colcaroli
347230e3c8
- Split backend/app/core into weather/, discord/, memory_kb/, search/ subpackages by responsibility, keeping config.py and llm.py as the shared foundation; updated all import sites accordingly. - Add local Markdown knowledge base support (local_knowledge.py, knowledge_base/fgo.md, LOCAL_KB_* settings) that embeds and injects relevant docs into the system prompt by topic similarity. - Add ruff and pyright as backend dev dependencies with basic config in pyproject.toml for linting and type checking. - Document local machine quirks in CLAUDE.md (Postgres on port 8080, SSLKEYLOGFILE breaking litellm/pyright via OPENSSL_Uplink crashes). - Add backend/run_dev.ps1 dev launcher and docs/discord-bot.md.
88 lines
3.3 KiB
Python
88 lines
3.3 KiB
Python
"""気象庁の無料JSON APIから天気予報を取得し、洗濯指数を判定する。"""
|
|
import logging
|
|
from datetime import datetime, timedelta
|
|
|
|
import httpx
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
JMA_FORECAST_URL = "https://www.jma.go.jp/bosai/forecast/data/forecast/{area_code}.json"
|
|
|
|
|
|
async def fetch_tomorrow_weather(area_code: str, area_name_keyword: str) -> dict:
|
|
"""指定エリアの明日の天気予報を取得する。
|
|
|
|
area_code: 気象庁の府県予報区コード（例: 千葉県 = 120000）
|
|
area_name_keyword: 府県内の地域区分名の一部（例: 市川市が属する「北西部」）
|
|
"""
|
|
async with httpx.AsyncClient(timeout=10.0) as client:
|
|
response = await client.get(JMA_FORECAST_URL.format(area_code=area_code))
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
|
|
# data[0] が短期予報（今日・明日・明後日）
|
|
short_term = data[0]
|
|
time_series = short_term["timeSeries"]
|
|
|
|
weather_series = time_series[0]
|
|
pop_series = time_series[1] if len(time_series) > 1 else None
|
|
|
|
tomorrow_index = _find_tomorrow_index(weather_series["timeDefines"])
|
|
|
|
weather_area = _find_area(weather_series["areas"], area_name_keyword)
|
|
weather_text = weather_area["weathers"][tomorrow_index] if weather_area else "不明"
|
|
|
|
pop_value = None
|
|
if pop_series:
|
|
pop_tomorrow_index = _find_tomorrow_index(pop_series["timeDefines"])
|
|
pop_area = _find_area(pop_series["areas"], area_name_keyword)
|
|
if pop_area and pop_area.get("pops"):
|
|
pops = pop_area["pops"]
|
|
# pops は3時間ごとの配列のことが多いため、明日に該当する値の最大値を取る
|
|
relevant_pops = [
|
|
int(p) for p, t in zip(pops, pop_series["timeDefines"])
|
|
if _is_same_date(t, pop_series["timeDefines"][pop_tomorrow_index]) and p
|
|
]
|
|
if relevant_pops:
|
|
pop_value = max(relevant_pops)
|
|
|
|
return {
|
|
"weather": weather_text.strip(),
|
|
"pop": pop_value,
|
|
}
|
|
|
|
|
|
def _find_tomorrow_index(time_defines: list[str]) -> int:
|
|
tomorrow = (datetime.now() + timedelta(days=1)).date()
|
|
for i, t in enumerate(time_defines):
|
|
dt = datetime.fromisoformat(t)
|
|
if dt.date() == tomorrow:
|
|
return i
|
|
return min(1, len(time_defines) - 1)
|
|
|
|
|
|
def _is_same_date(t1: str, t2: str) -> bool:
|
|
return datetime.fromisoformat(t1).date() == datetime.fromisoformat(t2).date()
|
|
|
|
|
|
def _find_area(areas: list[dict], keyword: str) -> dict | None:
|
|
for area in areas:
|
|
if keyword in area["area"]["name"]:
|
|
return area
|
|
return areas[0] if areas else None
|
|
|
|
|
|
def judge_laundry(weather: str, pop: int | None) -> tuple[str, str]:
|
|
"""天気と降水確率から洗濯指数を判定する。戻り値: (判定, 理由)"""
|
|
is_rainy = "雨" in weather or "雪" in weather
|
|
|
|
if is_rainy:
|
|
return "室内干し推奨", f"明日は「{weather}」のため"
|
|
if pop is None:
|
|
return "外干しOK（降水確率不明）", "降水確率データが取得できませんでした"
|
|
if pop <= 30:
|
|
return "外干しOK", f"降水確率{pop}%と低めです"
|
|
if pop <= 50:
|
|
return "室内干し推奨", f"降水確率{pop}%とやや高めです"
|
|
return "室内干し推奨", f"降水確率{pop}%と高いです"
|