使用Python编写一个每日天气提醒
目录
- 1. 引言
- 2. 系统设计与功能规划
- 2.1 系统架构设计
- 2.2 核心功能特性
- 2.3 技术选型
- 3. 准备工作与环境配置
- 3.1 申请天气API服务
- 3.2 安装必要的python库
- 3.3 环境变量配置
- 4. 核心模块实现
- 4.1 天气数据获取模块
- 4.2 智能分析与建议生成模块
- 4.3 个性化消息生成模块
- 4.4 消息发送模块
- 5. 系统集成与完整实现
- 6. 部署与运行
- 6.1 本地运行
- 6.2 服务器部署
- 6.3 微信版本特别说明
- 7. 个性化定制建议
- 7.1 添加特殊日期提醒
- 7.2 自定义关怀语句
- 7.3 天气相关的情话
- 8. 故障排除与优化
- 8.1 常见问题解决
- 8.2 性能优化建议
- 8.3 监控与维护
- 9. 完整代码文件
- 9.1 requirements.txt
- 9.2 weather_service.py
- 9.3 weather_analyzer.py
- 9.4 message_generator.py
- 9.5 message_sender.py
- 9.6 main.py
- 10. 总结
- 10.1 技术收获
- 10.2 情感价值
- 10.3 扩展思路
1. 引言
在这个快节奏的时代，一个小小的贴心举动往往能带来意想不到的温暖。想象一下，每天清晨，你的女朋友都能收到一条温馨的天气提醒，不仅包含准确的天气信息，还有你亲自编写的暖心提示——这样的关怀是不是很让人心动？
本文就将教你如何使用Python构建一个自动化的每日天气提醒系统。这个系统不仅实用，更重要的是它融入了你的个人情感，让技术变得有温度。无论你是编程新手还是有经验的开发者，都能通过本文学会如何将技术转化为贴心的浪漫工具。
2. 系统设计与功能规划
2.1 系统架构设计
在开始编码之前，让我们先来规划整个系统的架构。我们的天气提醒系统需要包含以下几个核心模块：
2.2 核心功能特性
我们的天气提醒系统将实现以下特色功能：
- 精准天气信息：获取实时天气、温度、湿度、风速等详细信息
- 智能穿衣建议：根据温度自动生成合适的穿衣建议
- 个性化关怀提示：结合天气情况提供贴心的生活建议
- 多平台支持：支持微信、邮件、短信等多种发送方式
- 定时发送：在指定时间自动发送提醒
- 异常处理：网络异常或服务不可用时的备用方案
2.3 技术选型
- 天气数据源：使用和风天气、OpenWeatherMap等免费API
- 消息发送：使用SMTP协议发送邮件，或itchat库发送微信
- 定时任务：使用APScheduler库管理定时任务
- 数据存储：使用SQLite存储配置和历史记录
- 配置管理：使用配置文件或环境变量管理敏感信息
3. 准备工作与环境配置
3.1 申请天气API服务
首先，我们需要选择一个天气数据提供商。这里以和风天气（免费版）为例：
- 访问和风天气官网注册账号
- 创建应用，获取API Key
- 了解API调用方式和返回数据格式
其他可选天气服务：
- OpenWeatherMap（免费版有限制）
- 心知天气
- 阿里云天气API
3.2 安装必要的Python库
pip install requests apscheduler python-dotenv sqlalchemy
如果需要微信支持，还需要安装：
pip install itchat-uos
3.3 环境变量配置
创建.env文件来存储敏感信息：
# 天气API配置 WEATHER_API_KEY=你的和风天气API_KEY WEATHER_CITY=北京 # 邮箱配置（用于发送邮件提醒） EMAIL_HOST=smtp.qq.com EMAIL_PORT=587 EMAIL_USER=你的邮箱@qq.com EMAIL_PASSWORD=你的邮箱授权码 # 接收方配置 RECIPIENT_EMAIL=女朋友的邮箱@gmail.com RECIPIENT_NAME=亲爱的
4. 核心模块实现
4.1 天气数据获取模块
import requests
import json
from datetime import datetime, timedelta
import logging
from typing import Dict, List, Optional
class WeatherService:
"""天气服务类，负责获取和处理天气数据"""
def __init__(self, api_key: str, city: str):
"""
初始化天气服务
Args:
api_key: 和风天气API Key
city: 城市名称
"""
self.api_key = api_key
self.city = city
self.base_url = "https://devapi.qweather.com/v7"
self.logger = logging.getLogger(__name__)
def get_city_location(self) -> Optional[Dict]:
"""获取城市位置信息（城市ID、坐标等）"""
url = f"{self.base_url}/weather/now"
params = {
'location': self.city,
'key': self.api_key
}
try:
response = requests.get(url, params=params, timeout=10)
response.raise_for_status()
data = response.json()
if data['code'] == '200':
# 从返回数据中提取位置信息
location_info = {
'city': data.get('location', [{}])[0].get('name', self.city),
'id': data.get('location', [{}])[0].get('id', ''),
'lat': data.get('location', [{}])[0].get('lat', ''),
'lon': data.get('location', [{}])[0].get('lon', '')
}
return location_info
else:
self.logger.error(f"获取位置信息失败: {data.get('message', '未知错误')}")
return None
except requests.exceptions.RequestException as e:
self.logger.error(f"网络请求失败: {str(e)}")
return None
except Exception as e:
self.logger.error(f"获取位置信息异常: {str(e)}")
return None
def get_current_weather(self) -> Optional[Dict]:
"""获取当前天气情况"""
url = f"{self.base_url}/weather/now"
params = {
'location': self.city,
'key': self.api_key,
'lang': 'zh' # 中文返回
}
try:
response = requests.get(url, params=params, timeout=10)
response.raise_for_status()
data = response.json()
if data['code'] == '200':
weather_data = data['now']
return {
'temp': weather_data.get('temp', 'N/A'), # 温度
'feels_like': weather_data.get('feelsLike', 'N/A'), # 体感温度
'text': weather_data.get('text', 'N/A'), # 天气状况文字
'wind_scale': weather_data.get('windScale', 'N/A'), # 风力等级
'wind_dir': weather_data.get('windDir', 'N/A'), # 风向
'humidity': weather_data.get('humidity', 'N/A'), # 湿度
'precip': weather_data.get('precip', '0'), # 降水量
'vis': weather_data.get('vis', 'N/A'), # 能见度
'pressure': weather_data.get('pressure', 'N/A'), # 气压
'update_time': weather_data.get('obsTime', '') # 更新时间
}
else:
self.logger.error(f"获取当前天气失败: {data.get('message', '未知错误')}")
return None
except requests.exceptions.RequestException as e:
self.logger.error(f"获取当前天气网络请求失败: {str(e)}")
return None
except Exception as e:
self.logger.error(f"获取当前天气异常: {str(e)}")
return None
def get_daily_forecast(self, days: int = 3) -> Optional[List[Dict]]:
"""获取多日天气预报"""
url = f"{self.base_url}/weather/{days}d"
params = {
'location': self.city,
'key': self.api_key,
'lang': 'zh'
}
try:
response = requests.get(url, params=params, timeout=10)
response.raise_for_status()
data = response.json()
if data['code'] == '200':
forecast_list = []
for day_data in data.get('daily', []):
forecast = {
'date': day_data.get('fxDate', ''),
'temp_max': day_data.get('tempMax', 'N/A'),
'temp_min': day_data.get('tempMin', 'N/A'),
'text_day': day_data.get('textDay', 'N/A'),
'text_night': day_data.get('textNight', 'N/A'),
'wind_scale_day': day_data.get('windScaleDay', 'N/A'),
'humidity': day_data.get('humidity', 'N/A'),
'precip': day_data.get('precip', '0'),
'uv_index': day_data.get('uvIndex', 'N/A')
}
forecast_list.append(forecast)
return forecast_list
else:
self.logger.error(f"获取天气预报失败: {data.get('message', '未知错误')}")
return None
except requests.exceptions.RequestException as e:
self.logger.error(f"获取天气预报网络请求失败: {str(e)}")
return None
except Exception as e:
self.logger.error(f"获取天气预报异常: {str(e)}")
return None
def get_weather_warning(self) -> Optional[List[Dict]]:
"""获取天气预警信息"""
url = f"{self.base_url}/warning/now"
params = {
'location': self.city,
'key': self.api_key,
'lang': 'zh'
}
try:
response = requests.get(url, params=params, timeout=10)
response.raise_for_status()
data = response.json()
if data['code'] == '200':
warnings = []
for warning in data.get('warning', []):
warning_info = {
'type': warning.get('type', ''),
'level': warning.get('level', ''),
'title': warning.get('title', ''),
'text': warning.get('text', '')
}
warnings.append(warning_info)
return warnings
else:
# 没有预警信息是正常情况
return []
except Exception as e:
phpself.logger.warning(f"获取天气预警信息失败: {str(e)}")
return []
4.2 智能分析与建议生成模块
class WeatherAnalyzer:
"""天气分析器，根据天气数据生成智能建议"""
def __init__(self):
self.advice_templates = {
'dressing': self._get_dressing_advice,
'umbrella': self._get_umbrella_advice,
'outdoor': self._get_outdoor_advice,
'health': self._get_health_advice,
'travel': self._get_travel_advice
}
def analyze_temperature(self, temp: int) -> Dict[str, str]:
"""分析温度并给出建议"""
temp = int(temp)
if temp >= 30:
level = "炎热"
feeling = "今天天气炎热，注意防暑降温"
elif temp >= 25:
level = "温暖"
feeling = "今天天气温暖，适合户外活动"
elif temp >= 15:
level = "舒适"
feeling = "今天天气舒适，是出行的好日子"
elif temp >= 5:
level = "凉爽"
feeling = "今天天气凉爽，记得适当添衣"
elif temp >= 0:
level = "寒冷"
feeling = "今天天气寒冷，注意保暖"
else:
level = "严寒"
feeling = "今天天气严寒，尽量减少外出"
return {'level': level, 'feeling': feeling}
def _get_dressing_advice(self, temp: int, weather_text: str) -> str:
"""生成穿衣建议"""
temp = int(temp)
if "雨" in weather_text:
rain_note = "有雨，建议穿防水的衣物"
else:
rain_note = ""
if temp >= 28:
return f" 轻薄夏装（短袖、裙子）{rain_note}"
elif temp >= 24:
return f" 夏季服装（短袖、薄长裤）{rain_note}"
elif temp >= 18:
return f" 春秋过渡装（长袖T恤、薄外套）{rain_note}"
elif temp >= 12:
return f" 春秋装（长袖、薄毛衣、外套）{rain_note}"
elif temp >= 6:
return f" 初冬装（毛衣、厚外套）{rain_note}"
elif temp >= 0:
return f" 冬装（羽绒服、厚毛衣）{rain_note}"
else:
return f" 厚冬装（加厚羽绒服、保暖内衣）{rain_note}"
def _get_umbrella_advice(self, weather_text: str, precip: float) -> str:
"""生成雨具建议"""
precip = float(precip)
if "雨" in weather_text or precip > 0:
if precip > 10:
return " 今天有大雨，一定要带伞，建议穿雨鞋"
elif precip > 5:
return "☔ 今天有中雨，记得带伞"
else:
return "️ 今天有小雨，建议带伞"
else:
return "☀️ 今天无雨，可以不用带伞"
def _get_outdoor_advice(self, weather_text: str, temp: int, wind_scale: str) -> str:
"""生成户外活动建议"""
temp = int(temp)
advice = []
# 温度适宜性
if 18 <= temp <= 26:
advice.append("今天的温度非常适合户外活动")
elif temp > 30:
advice.append("天气炎热，建议在阴凉处活动")
elif temp < 5:
advice.append("天气较冷，户外活动请注意保暖")
# 天气状况
if "晴" in weather_text:
advice.append("阳光很好，适合散步和运动")
elif "雨" in weather_text:
advice.append("有雨，建议室内活动")
elif "云" in weather_text:
advice.append("多云天气，适合各种户外活动")
elif "雪" in weather_text:
advice.append("下雪天，注意防滑")
# 风力影响
if int(wind_scale) >= 6:
advice.append("今天风较大，注意安全")
return "。".join(advice) if advice else "今天天气平稳，按计划活动即可"
def _get_health_advice(self, temp: int, humidity: int, uv_index: str) -> str:
"""生成健康建议"""
temp = int(temp)
humidity = int(humidity)
advice = []
# 温度相关
if temp >= 30:
advice.append("天气炎热，注意补充水分，预防中暑")
elif temp <= 0:
advice.append("天气寒冷，注意保暖，预防感冒")
# 湿度相关
if humidity >= 80:
advice.append("湿度较高，注意防潮")
elif humidity <= 30:
advice.append("空气干燥，注意保湿")
# 紫外线
if uv_index in ['5', '6', '7']:
advice.append("紫外线中等，建议使用防晒霜")
elif uv_index in ['8', '9', '10']:
advice.append("紫外线强，需要做好防晒措施")
elif uv_index in ['11', '12', '13', '14', '15']:
advice.append("紫外线很强，尽量避免长时间在阳光下")
return "。".join(advice) if advice else "今天天气条件对健康影响较小"
def _get_travel_advice(self, weather_text: str, vis: str) -> str:
"""生成出行建议"""
advice = []
if "雨" in weather_text:
advice.append("雨天路滑，开车请减速慢行")
elif "雪" in weather_text:
advice.append("雪天路滑，建议使用公共交通工具")
elif "雾" in weather_text:
advice.append("有雾，能见度较低，注意行车安全")
if vis and int(vis) < 1:
advice.append("能见度很低，出行请特别注意安全")
elif vis and int(vis) < 5:
advice.append("能见度一般，小心驾驶")
return "。".join(advice) if advice else "今天出行条件良好"
def generate_comprehensive_advice(self, weather_data: Dict) -> Dict[str, str]:
"""生成综合建议"""
temp = int(weather_data.get('temp', 20))
weather_text = weather_data.get('text', '')
precip = float(weather_data.get('precip', 0))
wind_scale = weather_data.get('wind_scale', '1')
humidity = int(weather_data.get('humidity', 50))
uv_index = weather_data.get('uv_index', '3')
vis = weather_data.get('vis', '10')
return {
'temperature_analysis': self.analyze_temperature(temp),
'dressing': self._get_dressing_advice(temp, weather_text),
'umbrella': self._get_umbrella_advice(weather_text, precip),
'outdoor': self._get_outdoor_advice(weather_text, temp, wind_scale),
'health': self._get_health_advice(temp, humidity, uv_index),
'travel': self._get_travel_advice(weather_text, vis)
}
4.3 个性化消息生成模块
import random
from datetime import datetime
class MessageGenerator:
"""消息生成器，创建个性化的天气提醒消息"""
def __init__(self, recipient_name: str):
self.recipient_name = recipient_name
self.greetings = [
"早安，{}！新的一天开始啦",
"早上好，{}！愿你有美好的一天",
"{}，醒来了吗？来看看今天的天气吧",
"嗨，{}！今天也要加油哦",
"{}，新的一天充满新的可能"
]
self.closing_remarks = [
"记得按时吃饭，爱你❤️",
"不管天气如何，我的关心永远不变",
"今天也要开开心心的哦",
"记得想我，就像我想你一样",
"愿你的一天像阳光一样灿烂✨"
]
self.special_date_remarks = {
'anniversary': "今天是我们的纪念日，晚上有惊喜哦",
'birthday': "生日快乐！我的小公主",
'valentine': "情人节快乐！我的爱人",
'weekend': "周末愉快！我们出去约会吧"
}
def _check_special_date(self) -> Optional[str]:
"""检查是否是特殊日期"""
today = datetime.now()
# 这里可以添加你们的特殊日期
special_dates = {
# 'anniversary': datetime(2024, 2, 14), # 纪念日示例
# 'birthday': datetime(2024, 3, 8), # 生日示例
}
for occasion, date in special_dates.items():
if today.month == date.month and today.day == date.day:
return occasion
# 检查周末
if today.weekday() >= 5: # 5=周六, 6=周日
return 'weekend'
return None
def generate_message(self, weather_data: Dict, advice_data: Dict,
OGxAmZh forecast_data: List[Dict] = None) -> str:
"""生成完整的天气提醒消息"""
# 问候语
greeting = random.choice(self.greetings).format(self.recipient_name)
# 当前天气摘要
current_temp = weather_data.get('temp', 'N/A')
weather_text = weather_data.get('text', 'N/A')
temp_analysis = advice_data['temperature_analysis']
weather_summary = (
f"️ 当前温度：{current_temp}C（{temp_analysis['level']}）\n"
f"️ 天气状况：{weather_text}\n"
f" 体感温度：{weather_data.get('feels_like', 'N/A')}C\n"
f" 湿度：{weather_data.get('humidity', 'N/A')}%\n"
f"️ 风力：{weather_data.get('wind_scale', 'N/A')}级 {weather_data.get('wind_dir', '')}"
)
# 生活建议
advice_section = (
f"\n 今日生活建议：\n"
f"{advice_data['dressing']}\n"
f"{advice_data['umbrella']}\n"
f"{advice_data['outdoor']}\n"
f"{advice_data['health']}\n"
f"{advice_data['travel']}"
)
# 天气预报
forecast_section = ""
if forecast_data:
forecast_section = "\n\n 未来三天预报：\n"
for i, forecast in enumerate(forecast_data[:3]):
date_str = forecast['date'][5:] # 只显示月日
day_name = "今天" if i == 0 else "明天" if i == 1 else "后天"
forecast_section += (
f"{day_name}({date_str})：{forecast['text_day']}，"
f"{forecast['temp_min']}~{forecast['temp_max']}C\n"
)
# 预警信息
warning_section = ""
warnings = weather_data.get('warnings', [])
if warnings:
warning_section = "\n⚠️ 天气预警：\n"
for warning in warnings:
warning_section += f"{warning['title']}：{warning['text']}\n"
# 特殊日期备注
special_remark = ""
special_occasion = self._check_special_date()
if special_occasion:
special_remark = f"\n {self.special_date_remarks[special_occasion]}"
# 结束语
closing = random.choice(self.closing_remarks)
# 组合所有部分
full_message = (
f"{greeting}\n\n"
f"{weather_summary}"
f"{advice_section}"
f"{forecast_section}"
f"{warning_section}"
f"{special_remark}\n\n"
f"{closing}"
)
return full_message
def generate_html_message(self, weather_data: Dict, advice_data: Dict,
forecast_data: List[Dict] = None) -> str:
"""生成HTML格式的消息（用于邮件）"""
# 这里实现HTML格式的消息生成
# 由于篇幅限制，具体实现略
pass
4.4 消息发送模块
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import logging
from typing import List
class MessageSender:
"""消息发送器，支持多种发送方式"""
def __init__(self):
self.logger = logging.getLogger(__name__)
def send_email(self, recipient: str, subject: str, content: str,
smtp_host: str, smtp_port: int,
username: str, password: str) -> bool:
"""通过邮件发送消息"""
try:
# 创建邮件对象
msg = MIMEMultipart()
msg['From'] = username
msg['To'] = recipient
msg['Subject'] = subject
# 添加文本内容
text_part = MIMEText(content, 'plain', 'utf-8')
msg.attach(text_part)
# 发送邮件
with smtplib.SMTP(smtp_host, smtp_port) as server:
server.starttls() # 启用TLS加密
server.login(username, password)
server.send_message(msg)
self.logger.info(f"邮件发送成功：{recipient}")
return True
except Exception as e:
self.logger.error(f"邮件发送失败：{str(e)}")
return False
def send_wechat(self, content: str) -> bool:
"""通过微信发送消息（需要安装itchat-uos）"""
try:
import itchat
# 查找指定备注的好友
friends = itchat.search_friends(name='女朋友的备注名') # 修改为实际备注
if friends:
friend = friends[0]
itchat.send(content, toUserName=friend['UserName'])
self.logger.info("微信消息发送成功")
return True
else:
self.logger.error("未找到指定的微信好友")
return False
except ImportError:
self.logger.error("未安装itchat-uos，无法发送微信消息")
return False
except Exception as e:
self.logger.error(f"微信消息发送失败：{str(e)}")
return False
def send_sms(self, phone_number: str, content: str,
api_key: str = None) -> bool:
"""通过短信API发送消息"""
# 这里可以实现短信发送功能
# 需要接入短信服务商API
self.logger.info("短信发送功能需要接入短信服务商API")
return False
def send_multiple_ways(self, content: str, config: Dict) -> bool:
"""尝试多种方式发送消息"""
success = False
# 尝试邮件发送
if config.get('email_enabled', False):
success = self.send_email(
recipient=config['recipient_email'],
subject=config.get('email_subject', '每日天气提醒'),
content=content,
smtp_host=config['smtp_host'],
smtp_port=config['smtp_port'],
username=config['email_username'],
password=config['email_password']
)
# 如果邮件发送失败，尝试微信
if not success and config.get('wechat_enabled', False):
success = self.send_wechat(content)
return success
5. 系统集成与完整实现
现在我们将所有模块整合在一起，创建一个完整的天气提醒系统：
#!/usr/bin/env python3
"""
每日天气提醒系统
为你的女朋友提供贴心的天气提醒服务
日期：2024年1月
"""
import os
import logging
from datetime import datetime
from dotenv import load_dotenv
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
import sqlite3
from typing import Dict, Any
# 导入自定义模块
from weather_service import WeatherService
from weather_analyzer import WeatherAnalyzer
from message_generator import MessageGenerator
from message_sender import MessageSender
# 加载环境变量
load_dotenv()
class DailyWeatherReminder:
"""每日天气提醒系统主类"""
def __init__(self, config: Dict[str, Any] = None):
"""初始化系统"""
# 配置日志
self._setup_logging()
# 加载配置
self.config = config or self._load_config()
# 初始化组件
self.weather_service = WeatherService(
api_key=self.config['weather_api_key'],
city=self.config['city']
)
self.weather_analyzer = WeatherAnalyzer()
self.message_generator = MessageGenerator(
recipient_name=self.config['recipient_name']
)
self.message_sender = MessageSender()
# 初始化数据库
self._init_database()
# 初始化调度器
self.scheduler = BackgroundScheduler()
self.logger.info("每日天气提醒系统初始化完成")
def _setup_logging(self):
"""配置日志系统"""
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('weather_reminder.log'),
logging.StreamHandler()
]
)
self.logger = logging.getLogger(__name__)
def _load_config(self) -> Dict[str, Any]:
"""从环境变量加载配置"""
return {
'weather_api_key': os.getenv('WEATHER_API_KEY'),
'city': os.getenv('WEATHER_CITY', '北京'),
'smtp_host': os.getenv('EMAIL_HOST'),
'smtp_port': int(os.getenv('EMAIL_PORT', 587)),
'email_username': os.getenv('EMAIL_USER'),
'email_password': os.getenv('EMAIL_PASSWORD'),
'recipient_email': os.getenv('RECIPIENT_EMAIL'),
'recipient_name': os.getenv('RECIPIENT_NAME', '亲爱的'),
'send_time': os.getenv('SEND_TIME', '07:30'), # 默认早上7:30发送
'email_enabled': True,
'wechat_enabled': False
}
def _init_database(self):
"""初始化SQLite数据库"""
self.conn = sqlite3.connect('weather_reminder.db', check_same_thread=False)
cursor = self.conn.cursor()
# 创建发送记录表
cursor.execute('''
CREATE TABLE IF NOT EXISTS send_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
send_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
success BOOLEAN,
message TEXT,
weather_data TEXT
)
''')
self.conn.commit()
def _log_send_attempt(self, success: bool, message: str, weather_data: Dict):
"""记录发送尝试"""
cursor = self.conn.cursor()
cursor.execute(
'INSERT INTO send_history (success, message, weather_data) VALUES (?, ?, ?)',
(success, message, str(weather_data))
)
self.conn.commit()
def get_weather_data(self) -> Dict[str, Any]:
"""获取完整的天气数据"""
self.logger.info("开始获取天气数据...")
weather_data = {}
# 获取当前天气
current_weather = self.weather_service.get_current_weather()
if current_weather:
weather_data.update(current_weather)
else:
self.logger.error("获取当前天气数据失败")
return {}
# 获取天气预报
forecast = self.weather_service.get_daily_forecast(days=3)
if forecast:
weather_data['forecast'] = forecast
else:
self.logger.warning("获取天气预编程客栈报数据失败")
# 获取天气预警
warnings = self.weather_service.get_weather_warning()
if warnings:
weather_data['warnings'] = warnings
self.logger.info("天气数据获取完成")
return weather_data
def generate_weather_report(self) -> str:
"""生成完整的天气报告"""
try:
# 获取天气数据
weather_data = self.get_weather_data()
if not weather_data:
return "抱歉，今天无法获取天气信息。请手动查看天气哦～"
# 生成智能建议
advice_data = self.weather_analyzer.generate_comprehensive_advice(weather_data)
# 生成个性化消息
forecast_data = weather_data.get('forecast', [])
message = self.message_generator.generate_message(
weather_data, advice_data, forecast_data
)
return message
except Exception as e:
self.logger.error(f"生成天气报告时发生错误: {str(e)}")
return f"今天天气服务出了点小问题，但我的关心不会变～记得查看天气哦！"
def send_daily_reminder(self):
"""发送每日提醒（主函数）"""
self.logger.info("开始发送每日天气提醒...")
try:
# 生成天气报告
message = self.generate_weather_report()
# 发送消息
success = self.message_sender.send_multiple_ways(message, self.config)
# 记录发送结果
weather_data = self.get_weather_data() # 重新获取用于记录
self._log_send_attempt(success, message, weather_data)
if success:
self.logger.info("每日天气提醒发送成功")
else:
self.logger.error("所有发送方式均失败")
return success
except Exception as e:
self.logger.error(f"发送每日提醒时发生错误: {str(e)}")
self._log_send_attempt(False, f"Error: {str(e)}", {})
return False
def setup_schedule(self):
"""设置定时任务"""
# 解析发送时间
hour, minute = map(int, self.config['send_time'].split(':'))
# 添加每日定时任务
self.scheduler.add_job(
self.send_daily_reminder,
trigger=CronTrigger(hour=hour, minute=minute),
id='daily_weather_reminder',
name='每日天气提醒',
replace_existing=True
)
self.logger.info(f"已设置每日 {hour:02d}:{minute:02d} 发送天气提醒")
def send_test_message(self):
"""发送测试消息"""
self.logger.info("发送测试消息...")
return self.send_daily_reminder()
def start(self):
"""启动系统"""
try:
# 设置定时任务
self.setup_schedule()
# 启动调度器
self.scheduler.start()
self.logger.info("天气提醒系统已启动")
# 立即发送一次测试消息
self.send_test_message()
# 保持程序运行
try:
while True:
# 主线程可以在这里执行其他任务
# 或者简单地等待
import time
time.sleep(60)
except KeyboardInterrupt:
self.logger.info("收到中断信号，正在关闭系统...")
except Exception as e:
self.logger.error(f"启动系统时发生错误: {str(e)}")
finally:
self.s编程客栈hutdown()
def shutdown(self):
"""关闭系统"""
if hasattr(self, 'scheduler') and self.scheduler.running:
self.scheduler.shutdown()
if hasattr(self, 'conn'):
self.conn.close()
self.logger.info("天气提醒系统已关闭")
def main():
"""主函数"""
# 可以在这里覆盖默认配置
custom_config = {
'city': '上海', # 修改为女朋友所在城市
'send_time': '08:00', # 修改为合适的发送时间
# 添加其他自定义配置...
}
# 创建并启动系统
reminder_system = DailyWeatherReminder(custom_config)
reminder_system.start()
if __name__ == "__main__":
main()
6. 部署与运行
6.1 本地运行
创建项目目录结构：
weather_reminder/
├── main.py├── weather_service.py├── weather_analyzer.py├── message_generator.py├── message_sender.py├── .env├── requirements.txt└── config.json (可选)
安装依赖：
pip install -r requirements.txt
配置环境变量（.env文件）：
WEATHER_API_KEY=你的和风天气API_KEY WEATHER_CITY=北京 EMAIL_HOST=smtp.qq.com EMAIL_PORT=587 EMAIL_USER=你的QQ邮箱 EMAIL_PASSWORD=QQ邮箱授权码 RECIPIENT_EMAIL=女朋友的邮箱 RECIPIENT_NAME=亲爱的 SEND_TIME=07:30
运行系统：
python main.py
6.2 服务器部署
对于24小时运行，建议部署到云服务器：
使用systemd服务（linux）：创建/etc/systemd/system/weather-reminder.service：
[Unit] Description=Daily Weather Reminder After=network.target [Service] Type=simple User=Ubuntu WorkingDirectory=/path/to/weather_reminder ExecStart=/usr/bin/python3 main.py Restart=always RestartSec=10 [Install] WantedBy=multi-user.target
使用docker部署：
创建Dockerfile：
FROM python:3.9-slim WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY . . CMD ["python", "main.py"]
6.3 微信版本特别说明
如果要使用微信版本，需要注意：
安装itchat-uos（解决微信网页版登录问题）：
pip install itchat-uos
修改代码使用微信发送：
# 在messagphpe_sender中启用微信发送
def send_wechat_message(self, content: str) -> bool:
import itchat
# 自动登录（会弹出二维码）
itchat.auto_login(hotReload=True, enableCmdQR=2)
# 发送给文件传输助手（测试用）
itchat.send(content, toUserName='filehelper')
# 或者发送给指定好友
friends = itchat.search_friends(name='女朋友的微信昵称')
if friends:
itchat.send(content, toUserName=friends[0]['UserName'])
itchat.logout()
return True
7. 个性化定制建议
7.1 添加特殊日期提醒
在MessageGenerator类中添加你们的特殊日期：
special_dates = {
'anniversary': datetime(2024, 5, 20), # 纪念日
'birthday': datetime(2024, 8, 15), # 生日
'valentine': datetime(2024, 2, 14), # 情人节
'first_meet': datetime(2024, 3, 8) # 初次见面日
}
7.2 自定义关怀语句
丰富问候语和结束语库：
self.greetings = [
"早安，我的小太阳{}！",
"{}宝贝，新的一天开始啦",
"嘿，{}！今天也是爱你的一天",
"早上好，{}！你是我今天第一个想念的人"
]
self.closing_remarks = [
"记得按时吃饭，你胃不好～",
"今天也要想我一千遍哦",
"工作再忙也要记得休息，心疼你",
"晚上视频哦，想你"
]
7.3 天气相关的情话
根据天气添加不同的情话：
def get_weather_pickup_line(self, weather_text: str) -> str:
"""根据天气生成情话"""
lines = {
'晴': "今天的阳光和你一样温暖我的心",
'雨': "雨水可以打湿街道，但打不湿我想你的心",
'云': "云朵遮住了太阳，但遮不住我对你的思念",
'雪': "雪花飘落的时候，我在想你有没有穿暖",
'风': "风儿吹过，带去我对你的牵挂"
}
for key, line in lines.items():
if key in weather_text:
return line
return "无论什么天气，你都是我的晴天"
8. 故障排除与优化
8.1 常见问题解决
问题1：API调用失败
- 检查API Key是否正确
- 确认网络连接正常
- 查看天气服务商的服务状态
问题2：邮件发送失败
- 检查邮箱授权码（不是登录密码）
- 确认SMTP服务器地址和端口正确
- 检查防火墙设置
问题3：定时任务不执行
- 确认系统时间正确
- 检查调度器是否正常启动
- 查看日志文件排查错误
8.2 性能优化建议
缓存城市信息：避免重复查询城市位置
错误重试机制：网络请求失败时自动重试
备用数据源：准备多个天气API备用
消息队列：使用Redis等管理发送任务
8.3 监控与维护
- 日志监控：定期检查系统日志
- API使用量：监控天气API调用次数
- 发送成功率：统计消息发送成功率
- 定期备份：备份配置和发送记录
9. 完整代码文件
由于完整的代码较长，这里提供各模块的文件划分：
9.1 requirements.txt
requests>=2.28.0 apscheduler>=3.9.0 python-dotenv>=0.19.0 python-dateutil>=2.8.0 itchat-uos>=1.5.0
9.2 weather_service.py
（包含WeatherService类，详见第4.1节）
9.3 weather_analyzer.py
（包含WeatherAnalyzer类，详见第4.2节）
9.4 message_generator.py
（包含MessageGenerator类，详见第4.3节）
9.5 message_sender.py
（包含MessageSender类，详见第4.4节）
9.6 main.py
（包含DailyWeatherReminder主类，详见第5节）
10. 总结
通过本文的指导，你已经学会了如何使用Python构建一个功能完整、贴心实用的每日天气提醒系统。这个系统不仅技术上有价值，更重要的是它承载了你的情感和关怀。
10.1 技术收获
- 掌握了API调用的最佳实践
- 学会了构建模块化的Python应用
- 了解了定时任务的实现方式
- 掌握了异常处理和日志记录
10.2 情感价值
这个天气提醒系统的真正价值在于：
- 持续关怀：每天准时的提醒体现持续的关注
- 个性化：根据天气提供的建议展现细心和体贴
- 实用性：真实的天气信息帮助日常生活
- 惊喜感：特殊日子的特别提醒增加浪漫氛围
10.3 扩展思路
你可以进一步扩展这个系统：
- 添加更多消息平台（钉钉、飞书等）
- 集成日历系统，提醒重要日程
- 添加机器学习，学习女朋友的偏好
- 开发手机APP，提供更好的交互体验
记住，技术只是工具，真正打动人心的永远是你的用心和关怀。希望这个天气提醒系统能为你们的感情增添一份科技的温暖！
温馨提示：在使用过程中，请尊重女朋友的隐私和偏好，确保她喜欢这样的关怀方式。技术应该服务于感情，而不是替代真实的交流和相处。祝你们的感情像精心编写的代码一样，稳定而美好！
以上就是使用Python编写一个每日天气提醒的详细内容，更多关于Python每日天气提醒的资料请关注编程客栈(www.devze.com)其它相关文章！
加载中,请稍侯......
精彩评论