天气查询工具包专业版

企业级天气数据平台,在免费版核心能力之上,扩展批量查询、历史数据、预警推送与多数据源聚合能力,适合物流、农业、旅游等需要规模化天气数据的企业与开发者。与免费版数据格式完全兼容,支持平滑升级。

概述

本工具包在免费版基础上,面向企业与专业开发者,提供多城市并发批量查询、历史天气回溯、极端天气预警推送与多数据源智能聚合能力。支持API封装、SDK集成、Webhook回调与定时任务,可无缝集成到企业业务系统。

核心能力

核心功能执行

用input_params参数进行配置。

输入: 用户提供核心功能执行所需的指令和必要参数。 处理: 按照skill规范执行核心功能执行操作,遵循单一意图原则。 输出: 返回核心功能执行的执行结果,包含操作状态和输出数据。

• 执行此能力时使用input_params参数,支持创建/查询/导出操作

参数配置与调用

用config_options参数进行配置。

输入: 用户提供参数配置与调用所需的指令和必要参数。 处理: 按照skill规范执行参数配置与调用操作,遵循单一意图原则。 输出: 返回参数配置与调用的执行结果,包含操作状态和输出数据。

• 执行此能力时使用config_options参数,支持修改/重置/导入操作

结果处理与输出

用output_format参数进行配置。

输入: 用户提供结果处理与输出所需的指令和必要参数。 处理: 按照skill规范执行结果处理与输出操作,遵循单一意图原则。 输出: 返回结果处理与输出的执行结果,包含操作状态和输出数据。

• 执行此能力时使用output_format参数,支持导出/保存/转换操作 能力覆盖范围：本skill的核心能力覆盖以下场景关键词：企业级天气数据平、支持批量查询、历史数据、预警推送与多数据、面向企业与开发者、的高级天气数据工、在免费版基础上扩、展批量查询、源聚合能力、核心能力、多城市批量天气查、询与并发处理、历史天气数据查询、极端天气预警与主、动推送通知、多数据源聚合与智、自定义天气数据缓、存与刷新策略、天气数据、封装与、天气可视化与报表、回调与定时任务集等。这些关键词对应description中声明的使用场景,均已在上述能力点中提供对应的操作支持。

使用场景

场景一:物流企业多点天气监控

# 示例
import concurrent.futures
import urllib.request
import json
from datetime import datetime

class WeatherBatchQuery:
    """批量天气查询器"""

    def __init__(self, max_workers=20):
        self.max_workers = max_workers
        self.cache = {}
        self.cache_ttl = 1800  # 30分钟缓存

    def query_single(self, city):
        """查询单个城市天气"""
        url = (
            f"https://api.open-meteo.com/v1/forecast"
            f"?latitude={city['lat']}&longitude={city['lon']}"
            f"¤t_weather=true&hourly=temperature_2m,precipitation"
        )
        try:
            with urllib.request.urlopen(url, timeout=10) as resp:
                data = json.loads(resp.read())
            return {
                "city": city["name"],
                "temperature": data["current_weather"]["temperature"],
                "windspeed": data["current_weather"]["windspeed"],
                "weathercode": data["current_weather"]["weathercode"],
                "status": "success"
            }
        except Exception as e:
            return {"city": city["name"], "status": "error", "message": str(e)}

    def batch_query(self, cities):
        """并发批量查询多个城市"""
        results = []
        with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            futures = {executor.submit(self.query_single, c): c for c in cities}
            for future in concurrent.futures.as_completed(futures):
                results.append(future.result())
        return results

# 使用示例:物流分拨中心天气监控
centers = [
    {"name": "北京分拨中心", "lat": 39.9, "lon": 116.4},
    {"name": "上海分拨中心", "lat": 31.2, "lon": 121.5},
    {"name": "广州分拨中心", "lat": 23.1, "lon": 113.3},
    {"name": "成都分拨中心", "lat": 30.6, "lon": 104.1},
    {"name": "武汉分拨中心", "lat": 30.5, "lon": 114.3},
]

query = WeatherBatchQuery(max_workers=10)
results = query.batch_query(centers)
for r in results:
    if r["status"] == "success":
        print(f"{r['city']}: {r['temperature']}°C, 风速{r['windspeed']}km/h")

场景二:历史天气数据回溯分析

# 历史天气查询命令示例
weather-pro history 
  --location "Beijing" 
  --start "2025-06-01" 
  --end "2025-06-30" 
  --metrics "temperature,precipitation,windspeed" 
  --format json 
  --output beijing_june.json

# 输出示例:
# [INFO] 查询历史天气: Beijing (2025-06-01 至 2025-06-30)
# [DATA] 数据源: Open-Meteo Archive API
# [STATS] 平均温度: 26.3°C
# [STATS] 最高温度: 38.2°C (2025-06-25)
# [STATS] 最低温度: 17.8°C (2025-06-03)
# [STATS] 累计降水: 78.5mm
# [EXPORT] 数据已导出: beijing_june.json

场景三:极端天气预警推送

# 天气预警引擎与Webhook推送
import json
import urllib.request

class WeatherAlertEngine:
    """天气预警引擎"""

    ALERT_THRESHOLDS = {
        "high_temp": 35,        # 高温预警阈值
        "low_temp": -10,        # 低温预警阈值
        "high_wind": 60,        # 大风预警阈值(km/h)
        "heavy_rain": 25,       # 暴雨预警阈值(mm/h)
        "storm_codes": [95, 96, 99]  # 雷暴天气代码
    }

    def check_alerts(self, weather_data, location):
        """检查天气数据是否触发预警"""
        alerts = []
        temp = weather_data.get("temperature", 0)
        wind = weather_data.get("windspeed", 0)
        code = weather_data.get("weathercode", 0)

        if temp >= self.ALERT_THRESHOLDS["high_temp"]:
            alerts.append({
                "type": "HIGH_TEMP",
                "level": "WARNING",
                "message": f"{location} 高温预警: {temp}°C",
                "action": "建议减少户外活动,注意防暑"
            })
        if wind >= self.ALERT_THRESHOLDS["high_wind"]:
            alerts.append({
                "type": "HIGH_WIND",
                "level": "WARNING",
                "message": f"{location} 大风预警: 风速{wind}km/h",
                "action": "建议固定室外物品,减少出行"
            })
        if code in self.ALERT_THRESHOLDS["storm_codes"]:
            alerts.append({
                "type": "STORM",
                "level": "URGENT",
                "message": f"{location} 雷暴天气预警",
                "action": "建议避免户外活动,远离高地"
            })
        return alerts

    def push_webhook(self, webhook_url, alert):
        """通过Webhook推送预警"""
        payload = json.dumps(alert).encode("utf-8")
        req = urllib.request.Request(
            webhook_url,
            data=payload,
            headers={"Content-Type": "application/json"},
            method="POST"
        )
        try:
            with urllib.request.urlopen(req, timeout=5) as resp:
                return resp.status == 200
        except Exception as e:
            print(f"推送失败: {e}")
            return False

# 使用示例
engine = WeatherAlertEngine()
weather = {"temperature": 38, "windspeed": 25, "weathercode": 95}
alerts = engine.check_alerts(weather, "北京分拨中心")
for alert in alerts:
    print(f"[{alert['level']}] {alert['message']}")
    # engine.push_webhook("https://hooks.example.com/weather", alert)

不适用场景

以下场景天气查询工具包专业版不适合处理：

• 逆向工程闭源API
• API安全渗透测试
• 非标准协议集成

触发条件

需要API集成、接口对接、Webhook配置、系统连接时使用。不适用于非本工具能力范围的需求。

快速开始

1. 配置数据源:在配置文件中指定主备数据源与切换策略
2. 导入城市列表:支持从CSV/YAML批量导入监控城市
3. 设置预警规则:定义温度、风速、降水等预警阈值
4. 配置推送通道:设置Webhook URL或邮件通知
5. 启动定时任务:配置查询频率与刷新策略
6. 查看聚合报表:通过可视化界面查看天气趋势

结果处理: 执行完成后,查看输出结果确认操作状态。成功时输出包含处理摘要和结果数据;失败时根据错误信息排查问题,查阅错误处理章节获取恢复步骤。

命令参数说明

• -Type: 命令参数,用于指定操作选项

配置示例

# 企业级天气工具包配置
weather_pro_config:
  edition: pro
  data_sources:
    primary: "open-meteo"
    fallback: "wttr"
    custom:
      - name: "企业气象服务"
        url: "https://weather.company.com/api"
        api_key_env: "COMPANY_WEATHER_KEY"
  batch_query:
    enabled: true
    max_concurrent: 50
    timeout: 10
    retry_count: 3
  history:
    enabled: true
    max_days: 365
    archive_source: "open-meteo-archive"
  alerts:
    enabled: true
    thresholds:
      high_temp: 35
      low_temp: -10
      high_wind: 60
      heavy_rain: 25
    channels:
      - type: "webhook"
        url: "https://hooks.example.com/weather"
      - type: "email"
        recipients: ["ops@company.com"]
  cache:
    enabled: true
    ttl: 1800
    storage: "redis"
  schedule:
    enabled: true
    interval: "*/30 * * * *"
    timezone: "Asia/Shanghai"
  export:
    formats: ["json", "csv", "pdf"]
    template_customization: true

最佳实践

• 数据源冗余:配置主备数据源,主源不可用时自动切换
• 合理缓存:根据业务需求设置TTL,平衡实时性与请求成本
• 预警分级:区分WARNING/URGENT/CRITICAL三级,对应不同推送策略
• 批量限流:并发查询控制在合理范围,避免被数据源限流
• 历史归档:定期导出历史数据到本地,便于长期趋势分析
• Webhook重试:推送失败时自动重试3次,间隔递增
• 监控告警:对工具本身进行监控,数据源不可用时及时通知

常见问题

Q1: 专业版与免费版的数据格式兼容吗?

完全兼容。专业版返回的JSON结构与免费版一致,仅在字段上做了扩展。原有解析代码无需修改即可正常工作,新增字段为可选。

Q2: 批量查询支持多少个城市?

单次批量查询支持最多500个城市,并发度可配置(默认50)。如需更大规模,建议分批处理或联系企业版定制。

Q3: 历史天气数据能查询多久以前?

Q4: 天气预警如何推送?

支持三种推送通道:

Q5: 多数据源如何智能切换?

# 数据源智能切换策略
class DataSourceManager:
    """多数据源管理器"""

    def __init__(self, sources):
        self.sources = sources  # 按优先级排序
        self.failure_count = {s: 0 for s in sources}

    def get_best_source(self):
        """选择最优数据源"""
        for source in self.sources:
            if self.failure_count[source] < 3:
                return source
        return self.sources[0]  # 全部失败时回退到首选

    def report_failure(self, source):
        """记录数据源失败"""
        self.failure_count[source] += 1

    def report_success(self, source):
        """记录数据源成功,重置失败计数"""
        self.failure_count[source] = 0

Q6: 如何集成到企业系统?

提供三种集成方式:

1. SDK集成:提供Python/Java/Go SDK,直接调用
2. REST API:封装为标准REST接口,支持HTTP调用
3. 消息队列:通过Kafka/RabbitMQ推送天气事件

# 依赖说明
pip install weather-pro-sdk

# API调用示例
python -c "
from weather_pro import WeatherClient
client = WeatherClient(api_key='your_key')
result = client.batch_query(['Beijing', 'Shanghai', 'Guangzhou'])
print(result)
"

Q7: 定时任务如何配置?

支持cron表达式配置定时查询,结果自动写入缓存或推送Webhook。

# 定时任务配置
schedule:
  - name: "每小时主城区天气"
    cron: "0 * * * *"
    cities: ["北京", "上海", "广州", "深圳"]
    action: "cache"
  - name: "极端天气预警检查"
    cron: "*/15 * * * *"
    cities: ["all_logistics_centers"]
    action: "alert"
    webhook: "https://hooks.example.com/weather"

依赖说明

运行环境

• Agent平台: 支持SKILL.md的任意AI Agent(Claude Code / Cursor / Codex / Gemini CLI等)
• 操作系统: Windows / macOS / Linux
• 网络连接: 需稳定访问天气数据源与推送通道
• 存储空间: 建议预留500MB用于历史数据缓存

第三方依赖

API Key 配置

• 基础功能:无需API Key,与免费版一致
• 企业气象服务(可选):需配置自定义API Key

  export COMPANY_WEATHER_KEY="your_company_weather_key"
  

• 短信推送(可选):需配置短信服务API Key

  export SMS_API_KEY="your_sms_api_key"
  

• Webhook:无需API Key,仅需配置URL

  export WEATHER_WEBHOOK_URL="https://hooks.example.com/weather"
  

可用性分类

• 分类: MD+EXEC(纯Markdown指令,部分功能需exec命令行执行)
• 说明: 基于Markdown的AI Skill,通过自然语言指令驱动Agent完成操作
• 专业版增强: 批量查询、历史数据、预警推送、多源聚合、API封装、定时任务
• 兼容性: 与免费版数据格式完全兼容,支持平滑升级
• 企业版扩展: 支持定制数据源、私有化部署、SLA保障(需联系销售)

错误处理

已知限制

• 需要API Key，无Key环境无法使用

示例

基本用法

输入：用户提供操作指令和必要参数

输出：返回执行结果,包含操作状态和输出数据

用户: 执行核心功能
Skill: 正在执行核心功能...
Skill: 执行完成,结果如下: 操作成功