Bedrockくんさぁ、毎朝天気教えてくれない？

はじめに

今回は毎朝天気を教えてくれるbotを作成しました。
前回の記事で作成したLINEbotを改修したものになっています。
よければ前回の記事もご覧ください。

天気を取得するには

天気の取得は公開されているAPIにリクエストを送ることで天気を取得する手法を取りました。
今回利用する候補となったのは以下の三つです。

Yahoo! Open Local Platform（YOLP）

YOLP は、Yahoo! が提供している天気・位置情報系のAPI群です。
天気予報APIも提供されており、短時間予報（いわゆる1時間おきの降水量予測）が取得できます。

ただし、このAPIで取得できるのは最大でも2時間先までの天気情報で、
たとえば「今日の最高気温」や「傘が必要かどうか」のような一日単位の天気を把握するのは難しいと感じました。

短時間予報やリアルタイムの降水情報には強いですが、今回の用途には合いませんでした。
今回は1日の天気等をまとめて持ってきたいと考えていたため、今回は利用しません。

気象庁ホームページ API

・公式に提供されていないAPIだが利用可能
・下記の防災エリアコード参考にリクエストを送ると、情報を取得できます。
https://www.jma.go.jp/bosai/common/const/area.json

例えば、東京都（エリアコード：130000）で実施すると、出力は以下のようになります。
例）https://www.jma.go.jp/bosai/forecast/data/overview_forecast/130000.json

{
  "publishingOffice": "気象庁",
  "reportDatetime": "2025-09-20T16:40:00+09:00",
  "targetArea": "東京都",
  "headlineText": "",
  "text": "　日本海には前線を伴った低気圧があって、東北東へ進んでいます。\n\n　東京地方は、雨や曇りとなっています。\n\n　２０日は、前線を伴った低気圧が日本海を東北東へ進み、湿った空気の影響を受ける見込みです。このため、雨夜遅く曇りで、雷を伴う所があるでしょう。伊豆諸島では雨で雷を伴い激しく降る所がある見込みです。\n\n　２１日は、はじめ前線が本州から伊豆諸島付近へ南下し、次第に高気圧に覆われますが、湿った空気の影響を受ける見込みです。このため、曇り朝から昼過ぎ晴れで、朝晩は雨や雷雨となる所があるでしょう。伊豆諸島では雨で雷を伴う所がある見込みです。\n\n【関東甲信地方】\n　関東甲信地方は、曇りや雨となっています。\n\n　２０日は、前線を伴った低気圧が日本海を東北東へ進み、湿った空気の影響を受ける見込みです。このため、曇りや雨で、雷を伴って激しく降る所があるでしょう。\n\n　２１日は、はじめ前線が本州から伊豆諸島付近へ南下し、次第に高気圧に覆われますが、湿った空気の影響を受ける見込みです。このため、曇りや晴れで、雨や雷雨となり、激しく降る所もあるでしょう。\n\n　関東地方と伊豆諸島の海上では、２０日から２１日にかけて波が高いでしょう。船舶は高波に注意してください。"
}

これも良さそうなのですが、この方法だとかなり大雑把なエリア単位でしか天気を取得できません...
「東京地方」という括りで取得しても、東京全体の平均的な天気が返ってくるので、東京のどこの天気か参考になりにくいです...

Free Weather API

最終的に採用したのが「Free Weather API」というサービスです。
このAPIでは、緯度・経度を指定して以下のような情報を取得できます：
• 今日の天気（晴れ、曇り、雨など）
• 最高・最低気温
• 降水確率
• 地点名（ジオコーディングも可）

市区町村単位で天気を取得できる上に、1日の天気情報をまとめて取れるのが非常に便利でした。

全体構成図

今回作成した仕組みは、「毎朝、指定した場所の天気を自動でLINEに通知する」ものです。
基本的な構成は、前回作成した記事をほぼ踏襲したものになっています。
主な変更点としては、Lambda関数の処理とEventBridgeの定期駆動の部分くらいかと思います。
構成図にある①〜④の処理を順番に説明すると以下のようになります
①EventBridge
→ EventBridgeのルールによって、毎朝決まった時間にLambdaが起動する。
②Lambda関数
→ Open-Meteo APIから「今日の天気」を取得（地点は環境変数で指定）
③Amazon Bedrock
→ 文章の自然な整形（整えたテキストだけを返すように制御）
④LINE Messaging API
→ 自分のLINEアカウントに通知メッセージをPush送信

Lambda関数を作成

実際のLambdaコードは以下のようになっています（コードは長めですが、ポイントごとで解説します）。

Lambdaコードをここに折りたたんで格納しています

import os
import json
import logging
import urllib.request
import urllib.parse
import datetime
import re
import boto3

logger = logging.getLogger()
logger.setLevel(logging.INFO)

# ===== 環境変数 =====
LINE_ACCESS_TOKEN = os.environ["LINE_ACCESS_TOKEN"]
LINE_TO           = os.environ["LINE_TO"]
PLACE_DISPLAY     = os.environ.get("PLACE_DISPLAY", "XXXX").strip()
HTTP_TIMEOUT      = float(os.environ.get("HTTP_TIMEOUT", "5"))


XXXX_LAT_ENV = os.environ.get("XXXX_LAT", "").strip()
XXXX_LON_ENV = os.environ.get("XXXX_LON", "").strip()


BEDROCK_REGION   = os.environ.get("BEDROCK_REGION")
BEDROCK_MODEL_ID = os.environ.get("BEDROCK_MODEL_ID")
bedrock = None
if BEDROCK_MODEL_ID:
    bedrock = boto3.client("bedrock-runtime", region_name=BEDROCK_REGION)

def _http(req: urllib.request.Request):
    with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT) as r:
        return r.getcode(), r.read().decode("utf-8", errors="replace")

def _get_json(url: str, headers: dict | None = None):
    req = urllib.request.Request(url, headers=headers or {})
    code, body = _http(req)
    if code != 200:
        raise RuntimeError(f"HTTP {code}: {body[:200]}")
    return json.loads(body)

def _push_line(to_id: str, text: str):
    data = {"to": to_id, "messages": [{"type": "text", "text": text[:4900]}]}
    req = urllib.request.Request(
        "https://api.line.me/v2/bot/message/push",
        data=json.dumps(data, ensure_ascii=False).encode("utf-8"),
        headers={"Content-Type": "application/json", "Authorization": f"Bearer {LINE_ACCESS_TOKEN}"},
    )
    code, body = _http(req)
    logger.info("LINE push status=%s body=%s", code, body[:300])

def _fetch_daily(lat: float, lon: float):
    jst = datetime.datetime.now(datetime.timezone(datetime.timedelta(hours=9)))
    start = end = jst.strftime("%Y-%m-%d")
    params = {
        "latitude": f"{lat}",
        "longitude": f"{lon}",
        "timezone": "Asia/Tokyo",
        "start_date": start,
        "end_date": end,
        "daily": ",".join([
            "weathercode",
            "temperature_2m_max",
            "temperature_2m_min",
            "precipitation_probability_max",
        ]),
    }
    url = "https://api.open-meteo.com/v1/forecast?" + urllib.parse.urlencode(params)
    data = _get_json(url, {"User-Agent": "lambda-openmeteo-XXXX/2.1"})
    daily = data.get("daily") or {}

    def _first(lst, default=None):
        try:
            return lst[0]
        except Exception:
            return default

    return {
        "weathercode": _first(daily.get("weathercode")),
        "tmax": _first(daily.get("temperature_2m_max")),
        "tmin": _first(daily.get("temperature_2m_min")),
        "pop": _first(daily.get("precipitation_probability_max")),  # %
    }

# ===== WMO weathercode → 日本語 =====
_WMO_JA = {
    0: "快晴", 1: "晴れがち", 2: "薄曇り", 3: "くもり",
    45: "霧", 48: "樹氷着霧",
    51: "弱い霧雨", 53: "やや強い霧雨", 55: "強い霧雨",
    56: "弱い凍雨", 57: "強い凍雨",
    61: "弱い雨", 63: "やや強い雨", 65: "強い雨",
    66: "弱い凍雨", 67: "強い凍雨",
    71: "弱い雪", 73: "やや強い雪", 75: "強い雪", 77: "雪片",
    80: "にわか雨（弱）", 81: "にわか雨（中）", 82: "にわか雨（強）",
    85: "にわか雪（弱）", 86: "にわか雪（強）",
    95: "雷雨の可能性", 96: "雷雨（弱いひょう）", 99: "雷雨（強いひょう）",
}
def _wmo_to_ja(code):
    if code is None:
        return "不明"
    return _WMO_JA.get(int(code), f"天気コード{code}")

def _umbrella_and_comment(weathercode: int | None, pop: float | None, tmax: float | None, tmin: float | None):
    """
    傘: 必携 / 折りたたみ推奨 / 不要
    一言: 天気・降水・気温の要点を短い日本語で2文以内
    """
    code = int(weathercode) if weathercode is not None else None
    p = float(pop) if pop is not None else None

    must_codes = {63, 65, 80, 81, 82, 95, 96, 99}  # 強めの雨・雷雨系
    fold_codes = {51, 53, 55, 61, 71, 80, 85}      # 霧雨・弱雨・にわか系（弱）

    if (p is not None and p >= 70) or (code in must_codes):
        umbrella = "必携"
        umbrella_mark = "☔"
    elif (p is not None and p >= 40) or (code in fold_codes):
        umbrella = "折りたたみ推奨"
        umbrella_mark = "🌂"
    else:
        umbrella = "不要"
        umbrella_mark = "—"

    comments = []

   
    if code in {95,96,99}:
        comments.append("雷雨に注意")
    elif p is not None and p >= 70:
        comments.append("雨がちで足元注意")
    elif p is not None and 40 <= p < 70:
        comments.append("にわか雨に気をつけて")
    elif code in {0,1}:
        comments.append("日差しありUV対策を")
    elif code in {2,3}:
        comments.append("雲が多くスッキリしない")

    
    if tmax is not None and tmax >= 30:
        comments.append("暑さ対策を")
    elif tmax is not None and tmax <= 10:
        comments.append("暖かくして")
    if tmax is not None and tmin is not None:
        try:
            if (tmax - tmin) >= 8:
                comments.append("寒暖差に注意")
        except Exception:
            pass

    one_liner = "・".join(comments[:2]) if comments else ""

    return umbrella, umbrella_mark, one_liner

def _bedrock_polish(text: str) -> str:
    if not bedrock or not BEDROCK_MODEL_ID:
        return text

    prompt = (
        "次の日本語テキストを、数字や意味を変えずに簡潔で自然に整えてください。"
        "出力は本文のみ。『以下は整えたテキストです』などの前置きやラベル、説明文は付けないでください。"
        "改行は維持して2文まで。絵文字はそのまま保持してください。\n\n"
        + text
    )

    body = {
        "anthropic_version": "bedrock-2023-05-31",
        "max_tokens": 200,
        "messages": [{"role": "user", "content": [{"type": "text", "text": prompt}]}],
    }
    try:
        res = bedrock.invoke_model(
            modelId=BEDROCK_MODEL_ID,
            body=json.dumps(body),
            contentType="application/json",
            accept="application/json",
        )
        payload = json.loads(res["body"].read())
        out = (payload.get("content") or [{}])[0].get("text", "").strip()

        out = _strip_bedrock_preamble(out)
        return out or text
    except Exception as e:
        logger.exception("Bedrock polish failed: %s", e)
        return text

def _strip_bedrock_preamble(s: str) -> str:
    if not s:
        return s

    patterns = [
        r"^\s*以下は整えたテキストです\s*[:：、。\-]*\s*",
        r"^\s*以下が整えたテキストです\s*[:：、。\-]*\s*",
        r"^\s*整えたテキスト\s*[:：、。\-]*\s*",
        r"^\s*出力\s*[:：、。\-]*\s*",
    ]
    for pat in patterns:
        s = re.sub(pat, "", s, flags=re.IGNORECASE)
    s = s.strip()
    if s.startswith("```") and s.endswith("```"):
        s = s.strip("`").strip()
    s = re.sub(r'^[「『"]\s*', "", s)
    s = re.sub(r'\s*[」』"]$', "", s)

    return s.strip()

def _compose_message(place: str, d: dict) -> str:
    desc = _wmo_to_ja(d.get("weathercode"))
    tmax = d.get("tmax"); tmin = d.get("tmin"); pop = d.get("pop")
    tmax_s = f"{int(round(tmax))}℃" if tmax is not None else "—℃"
    tmin_s = f"{int(round(tmin))}℃" if tmin is not None else "—℃"
    pop_s  = f"{int(round(pop))}%" if pop is not None else "—"

    # 傘 & コメント
    umbrella, umb_mark, one = _umbrella_and_comment(d.get("weathercode"), pop, tmax, tmin)

    line1 = f"{place}の天気は、{desc}。最高{tmax_s}・最低{tmin_s}、降水確率{pop_s}です。"
    line2 = f"傘: {umbrella} {umb_mark}／ひと言: {one}" if one else f"傘: {umbrella} {umb_mark}"

    text = line1 + "\n" + line2
    return _bedrock_polish(text)

def lambda_handler(event, context):
    try:
        lat = float(XXXX_LAT_ENV)
        lon = float(XXXX_LON_ENV)
    except Exception:
        msg = (
            f"{PLACE_DISPLAY}の天気は、位置情報（XXXX_LAT / XXXX_LON）の設定が不足しています。"
            " 管理者に問い合わせてください。"
        )
        _push_line(LINE_TO, msg)
        return {"ok": False, "reason": "latlon_env_missing"}

    try:
        daily = _fetch_daily(lat, lon)
    except Exception as e:
        logger.exception("Open-Meteo fetch failed: %s", e)
        _push_line(LINE_TO, f"{PLACE_DISPLAY}の天気は、現在取得できませんでした。")
        return {"ok": False, "reason": "fetch_failed"}

    msg = _compose_message(PLACE_DISPLAY, daily)
    _push_line(LINE_TO, msg)

    return {"ok": True, "place": PLACE_DISPLAY, "lat": lat, "lon": lon, "pushed": True}

処理の解説

コードの主な処理は以下の通りです。

環境変数の読み取り

LINE_ACCESS_TOKEN = os.environ["LINE_ACCESS_TOKEN"]
LINE_TO           = os.environ["LINE_TO"]
....
XXXX_LAT_ENV = os.environ.get("XXXX_LAT", "").strip()
XXXX_LON_ENV = os.environ.get("XXXX_LON", "").strip()

・LINEの認証情報や、通知先のユーザーID、位置情報（緯度経度）などは環境変数で管理しています。
・また、ところどころ「XXXX」という文字列が出てきますが、これは私が住んでいる場所を変数名にしているのに気づき急遽執筆時変更しました...汗

Open-Meteo APIで天気取得

def _fetch_daily(lat: float, lon: float):
    ...

・Open-Meteo API から「今日」の天気情報を取得します
・weathercode（天気コード）や tmax（最高気温）などを返す形式に変換しています。

傘の必要性とコメントの判定

def _umbrella_and_comment(weathercode, pop, tmax, tmin):
    ...

・天気コード・降水確率・気温から、傘の必要性やコメントを判定します。

メッセージの整形（自然な文章）

def _compose_message(place: str, d: dict) -> str:
    ...

・文章は「地名＋天気＋気温＋降水確率」と「傘＋コメント」を組み合わせた2行構成にしています。
・weathercode（天気コード）や tmax（最高気温）などを返す形式に変換しています。

数日間動作確認していたところ、プロンプトが甘かったのか以下のようにメッセージ冒頭に「整えたテキストは以下の通りです：」と出てくるようになってしまいました。

確かにBedrockでやっている処理ですが、必要ない情報なので、プロンプトを追加してこの情報が出ないように抑制しています。

prompt = (
        "次の日本語テキストを、数字や意味を変えずに簡潔で自然に整えてください。"
        "出力は本文のみ。『以下は整えたテキストです』などの前置きやラベル、説明文は付けないでください。"
        "改行は維持して2文まで。絵文字はそのまま保持してください。\n\n"
        + text
    )

EventBridgeの設定

毎朝通知を受け取りたいので、EventBridgeで「ルール（Rule）」を作成して、定期実行のスケジュールを定義します。
スケジュールは cron式 または rate式 で指定可能ですが、今回は「毎日8時」にしたいので cron式を使います。
例えば、以下のような cron 式を使えばOKです

設定後、「次の10個のトリガー日」が表示されますので、その後ターゲットを作成したLambda関数に指定し、他に問題なければスケジュールを設定します。

これで毎朝動くようになってくれるはずです

動作確認

一通り設定が終わったら、ちゃんと動くか確認しておきます。

きちんと朝8時に通知が来ており、不要なコメントも含まれていませんね。
ちなみに今も毎日動いて伝えてくれています笑

まとめ

今回紹介した構成では、以下の仕組みを通して
毎朝、自動でその日の天気をLINEに通知することができるようになりました。

• 緯度経度に基づいたピンポイントな天気情報を取得（Open-Meteo API）
• 天気コード・気温・降水確率に応じて傘の必要性や一言コメントを自動判定
• Amazon Bedrockを使って、自然で読みやすい文章に整形
• EventBridgeで毎朝自動実行

また、変数名に自分の住所をがっつり入れていたのは本当に危なかったです。
記事公開直前に気付いて良かったとしか言えない...

「Bedrockでの自然文整形」は、通知内容の読みやすさを大きく、楽に向上させるポイントでした。
今後も生成AIを活用して、ちょっとした便利ツールを考えていこうと思います。