将商业核心数据源 URL 从源码迁移到环境变量,VPS .env 已补全

This commit is contained in:
2569718930@qq.com
2026-05-19 15:42:03 +08:00
parent 7e0d955d22
commit 3b8098051e
11 changed files with 57 additions and 33 deletions
+5 -2
View File
@@ -1,5 +1,6 @@
from __future__ import annotations
import os
import time
from datetime import datetime
from typing import Any, Dict, List, Optional
@@ -8,6 +9,8 @@ from loguru import logger
from src.utils.metrics import record_source_call
_JMA_BASE = os.getenv("JMA_AMEDAS_BASE_URL", "").strip() or "https://www.jma.go.jp"
JMA_AMEDAS_STATIONS: Dict[str, Dict[str, Any]] = {
"tokyo": {
@@ -60,13 +63,13 @@ class JmaAmedasSourceMixin:
try:
latest_time_text = self._jma_http_get_text(
"https://www.jma.go.jp/bosai/amedas/data/latest_time.txt"
f"{_JMA_BASE}/bosai/amedas/data/latest_time.txt"
).strip()
latest_dt = datetime.fromisoformat(latest_time_text)
bucket_hour = (latest_dt.hour // 3) * 3
bucket_key = f"{latest_dt.strftime('%Y%m%d')}_{bucket_hour:02d}"
station_code = str(meta.get("station_code") or "").strip()
url = f"https://www.jma.go.jp/bosai/amedas/data/point/{station_code}/{bucket_key}.json"
url = f"{_JMA_BASE}/bosai/amedas/data/point/{station_code}/{bucket_key}.json"
getter = getattr(self, "_http_get_json", None)
if callable(getter):