使用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 扩展思路

在这个快节奏的时代，一个小小的贴心举动往往能带来意想不到的温暖。想象一下，每天清晨，你的女朋友都能收到一条温馨的天气提醒，不仅包含准确的天气信息，还有你亲自编写的暖心提示——这样的关怀是不是很让人心动？

本文就将教你如何使用Python构建一个自动化的每日天气提醒系统。这个系统不仅实用，更重要的是它融入了你的个人情感，让技术变得有温度。无论你是编程新手还是有经验的开发者，都能通过本文学会如何将技术转化为贴心的浪漫工具。

在开始编码之前，让我们先来规划整个系统的架构。我们的天气提醒系统需要包含以下几个核心模块：

我们的天气提醒系统将实现以下特色功能：

• 精准天气信息：获取实时天气、温度、湿度、风速等详细信息
• 智能穿衣建议：根据温度自动生成合适的穿衣建议
• 个性化关怀提示：结合天气情况提供贴心的生活建议
• 多平台支持：支持微信、邮件、短信等多种发送方式
• 定时发送：在指定时间自动发送提醒
• 异常处理：网络异常或服务不可用时的备用方案

• 天气数据源：使用和风天气、OpenWeatherMap等免费API
• 消息发送：使用SMTP协议发送邮件，或itchat库发送微信
• 定时任务：使用APScheduler库管理定时任务
• 数据存储：使用SQLite存储配置和历史记录
• 配置管理：使用配置文件或环境变量管理敏感信息

首先，我们需要选择一个天气数据提供商。这里以和风天气（免费版）为例：

• 访问和风天气官网注册账号
• 创建应用，获取API Key
• 了解API调用方式和返回数据格式

其他可选天气服务：

• OpenWeatherMap（免费版有限制）
• 心知天气
• 阿里云天气API

pip install requests apscheduler python-dotenv sqlalchemy

如果需要微信支持，还需要安装：

pip install itchat-uos

创建.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=亲爱的

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:
            self.logger.warning(f"获取天气预警信息失败: {str(e)}")
            return []

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)
        }

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, 
                        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')}°Cn"
            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 = "nn📅 未来三天预报：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']}°Cn"
                )
        
        # 预警信息
        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}nn"
            f"{weather_summary}"
            f"{advice_section}"
            f"{forecast_section}"
            f"{warning_section}"
            f"{special_remark}nn"
            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

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

现在我们将所有模块整合在一起，创建一个完整的天气提醒系统：

#!/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.shutdown()
    
    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()

创建项目目录结构：

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

对于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"]

如果要使用微信版本，需要注意：

安装itchat-uos（解决微信网页版登录问题）：

pip install itchat-uos

修改代码使用微信发送：

# 在message_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

在MessageGenerator类中添加你们的特殊日期：

special_dates = {
    'anniversary': datetime(2024, 5, 20),  # 纪念日
    'birthday': datetime(2024, 8, 15),     # 生日
    'valentine': datetime(2024, 2, 14),    # 情人节
    'first_meet': datetime(2024, 3, 8)     # 初次见面日
}

丰富问候语和结束语库：

self.greetings = [
    "早安，我的小太阳{}！",
    "{}宝贝，新的一天开始啦",
    "嘿，{}！今天也是爱你的一天",
    "早上好，{}！你是我今天第一个想念的人"
]

self.closing_remarks = [
    "记得按时吃饭，你胃不好～",
    "今天也要想我一千遍哦",
    "工作再忙也要记得休息，心疼你",
    "晚上视频哦，想你😘"
]

根据天气添加不同的情话：

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 "无论什么天气，你都是我的晴天"

问题1：API调用失败

• 检查API Key是否正确
• 确认网络连接正常
• 查看天气服务商的服务状态

问题2：邮件发送失败

• 检查邮箱授权码（不是登录密码）
• 确认SMTP服务器地址和端口正确
• 检查防火墙设置

问题3：定时任务不执行

• 确认系统时间正确
• 检查调度器是否正常启动
• 查看日志文件排查错误

缓存城市信息：避免重复查询城市位置

错误重试机制：网络请求失败时自动重试

备用数据源：准备多个天气API备用

消息队列：使用Redis等管理发送任务

• 日志监控：定期检查系统日志
• API使用量：监控天气API调用次数
• 发送成功率：统计消息发送成功率
• 定期备份：备份配置和发送记录

由于完整的代码较长，这里提供各模块的文件划分：

requests>=2.28.0
apscheduler>=3.9.0
python-dotenv>=0.19.0
python-dateutil>=2.8.0
itchat-uos>=1.5.0

（包含WeatherService类，详见第4.1节）

（包含WeatherAnalyzer类，详见第4.2节）

（包含MessageGenerator类，详见第4.3节）

（包含MessageSender类，详见第4.4节）

（包含DailyWeatherReminder主类，详见第5节）

通过本文的指导，你已经学会了如何使用Python构建一个功能完整、贴心实用的每日天气提醒系统。这个系统不仅技术上有价值，更重要的是它承载了你的情感和关怀。

• 掌握了API调用的最佳实践
• 学会了构建模块化的Python应用
• 了解了定时任务的实现方式
• 掌握了异常处理和日志记录

这个天气提醒系统的真正价值在于：

• 持续关怀：每天准时的提醒体现持续的关注
• 个性化：根据天气提供的建议展现细心和体贴
• 实用性：真实的天气信息帮助日常生活
• 惊喜感：特殊日子的特别提醒增加浪漫氛围

你可以进一步扩展这个系统：

• 添加更多消息平台（钉钉、飞书等）
• 集成日历系统，提醒重要日程
• 添加机器学习，学习女朋友的偏好
• 开发手机APP，提供更好的交互体验

记住，技术只是工具，真正打动人心的永远是你的用心和关怀。希望这个天气提醒系统能为你们的感情增添一份科技的温暖！

温馨提示：在使用过程中，请尊重女朋友的隐私和偏好，确保她喜欢这样的关怀方式。技术应该服务于感情，而不是替代真实的交流和相处。祝你们的感情像精心编写的代码一样，稳定而美好！

以上就是使用Python编写一个每日天气提醒的详细内容，更多关于Python每日天气提醒的资料请关注风君子博客其它相关文章！

您可能感兴趣的文章:

• 基于Python实现智能天气提醒助手
• Python爬虫之获取心知天气API实时天气数据并弹窗提醒
• python编写朋克风格的天气查询程序
• Python实战之天气预报系统的实现