基于Python实现智能天气提醒助手_Python

项目概述

今天分享一个实用的python天气提醒助手开发方案，这个工具可以方便地集成到青龙面板或其他调度框架中使用。助手能获取实时天气和24小时预报，并通过ai生成贴心的生活建议，最后推送通知给用户。

核心功能

实时天气查询 - 通过阿里云api获取当前天气数据

天气预报获取 - 查询24小时天气变化情况

ai智能建议 - 基于天气数据生成人文关怀建议

消息推送 - 通过bark服务发送通知到ios设备

技术实现

1. 天气api集成

def now_weather():
    try:
        host = 'https://ali-weather.showapi.com'
        path = '/hour24'
        appcode = os.getenv("appcode")
        querys = 'area=xxx'
        url = host + path + '?' + querys
        headers = {'authorization': 'appcode ' + appcode}
        resp = requests.get(url, headers=headers)
        resp.raise_for_status()
        return resp.json()

2. ai建议生成

def get_ai_advice(prompt):
    try:
        client = openai(
            base_url=os.getenv("baseurl"),
            api_key=os.getenv("apikey")
        )
        response = client.chat.completions.create(
            model=os.getenv("model"),
            messages=[
                {"role": "assistant", "content": "你是一个天气助手..."},
                {"role": "user", "content": prompt},
            ],
        )
        return response.choices[0].message.content

3. 消息推送

def bark(title: str, content: str) -> none:
    bark_push = os.getenv("bark_push")
    url = f'https://api.day.app/{bark_push}' if not bark_push.startswith("http") else bark_push
    data = {"title": title, "body": content}
    response = requests.post(url, data=json.dumps(data), headers=headers, timeout=15).json()

环境配置

使用时需要配置以下环境变量：

appcode        # 阿里云api的appcode
baseurl        # openai api地址
apikey         # openai api密钥
model          # 使用的ai模型
bark_push      # bark推送地址或设备码

使用方法

if __name__ == "__main__":
    advice = main()  # 获取天气建议
    bark("天气情况", advice)  # 推送通知

完整代码

import logging
logging.basicconfig(level=logging.info,format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
# 配置日志
logger = logging.getlogger('weather')
import requests, json, os, re
from datetime import datetime
from openai import openai

def now_weather():
    try:
        host = 'https://ali-weather.showapi.com'
        path = '/hour24'
        appcode = os.getenv("appcode")
        querys = 'area=xxx'
        url = host + path + '?' + querys
        headers = {
            'authorization': 'appcode ' + appcode
        }
        resp = requests.get(url, headers=headers)
        resp.raise_for_status()
        return resp.json()
    except exception as e:
        log.error(f"获取当前天气失败: {str(e)}")
        return none

def get_hour24():
    try:
        host = 'https://ali-weather.showapi.com'
        path = '/hour24'
        appcode = os.getenv("appcode")
        querys = 'area=xxx'
        url = host + path + '?' + querys
        headers = {
            'authorization': 'appcode ' + appcode
        }
        resp = requests.get(url, headers=headers)
        resp.raise_for_status()
        return resp.json()
    except exception as e:
        log.error(f"获取24小时天气失败: {str(e)}")
        return none
   
def current_time():
    datiem = datetime.now()
    return datiem.strftime("%y-%m-%d %h:%m:%s")

def get_ai_advice(prompt):
    try:
        client = openai(
            base_url=os.getenv("baseurl"),  # 使用配置的base_url
            api_key=os.getenv("apikey")  # 使用配置的api_key
        )
        response = client.chat.completions.create(
            model=os.getenv("model"),
            messages=[
                {"role": "assistant", "content": "你是一个天气助手，根据实际数据，给用户精准的天气数据总结和人文化的生活建议。例如:明天天气小雨，和今天温差较大，注意保暖，同时伴有大风，注意安全，睡觉前记得关紧门窗。"},
                {
                    "role": "user",
                    "content": prompt,
                },
            ],
        )
        return response.choices[0].message.content
    except exception as e:
        log.error(f"获取ai建议失败: {str(e)}")
        return "无法获取ai建议"


def main():
    current_weather = now_weather()
    forecast_weather = get_hour24()
    
    promot = f"""
    当前时间是：{current_time()}
    当前天气情况：{current_weather}
    24小时天气预报：{forecast_weather}
    注意：
    1. 请根据实际数据，给用户精准的天气数据总结和人文化的生活建议。
    2. 请不要使用markdown格式输出。尽量口语化
    """
    print(promot)
    return get_ai_advice(promot)


push_config = {
    'bark_push': '',                    # bark ip 或设备码，例：https://api.day.app/dxhcxxxxxrxxxxxxcm/
    'bark_archive': '',                 # bark 推送是否存档
    'bark_group': '',                   # bark 推送分组
    'bark_sound': '',                   # bark 推送声音
    'bark_icon': '',                    # bark 推送图标
    'bark_level': '',                   # bark 推送时效性
    'bark_url': '',                     # bark 推送跳转url
}


def bark(title: str, content: str) -> none:
    """
    使用 bark 推送消息。
    """
    bark_push = os.getenv("bark_push")
    if not bark_push:
        print("bark 服务的 bark_push 未设置!!\n取消推送")
        return
    print("bark 服务启动")

    if bark_push.startswith("http"):
        url = f'{bark_push}'
    else:
        url = f'https://api.day.app/{bark_push}'

    bark_params = {
        "bark_archive": "isarchive",
        "bark_group": "group",
        "bark_sound": "sound",
        "bark_icon": "icon",
        "bark_level": "level",
        "bark_url": "url",
    }
    data = {
        "title": title,
        "body": content,
    }
    for pair in filter(
        lambda pairs: pairs[0].startswith("bark_")
        and pairs[0] != "bark_push"
        and pairs[1]
        and bark_params.get(pairs[0]),
        push_config.items(),
    ):
        data[bark_params.get(pair[0])] = pair[1]
    headers = {"content-type": "application/json;charset=utf-8"}
    response = requests.post(
        url=url, data=json.dumps(data), headers=headers, timeout=15
    ).json()

    if response["code"] == 200:
        print("bark 推送成功！")
    else:
        print("bark 推送失败！")


advice = main()
print(advice)
bark("天气情况",advice)