Spaces:
Sleeping
Sleeping
Update cwa_service.py
Browse files- cwa_service.py +15 -6
cwa_service.py
CHANGED
|
@@ -28,6 +28,7 @@ def _parse_cwa_time(s: str) -> tuple[str, str]:
|
|
| 28 |
|
| 29 |
# --- 地震預警 (CWA_ALARM_API) ---
|
| 30 |
def fetch_cwa_alarm_list(limit: int = 5) -> str:
|
|
|
|
| 31 |
try:
|
| 32 |
r = requests.get(CWA_ALARM_API, timeout=10)
|
| 33 |
r.raise_for_status()
|
|
@@ -51,12 +52,22 @@ def fetch_cwa_alarm_list(limit: int = 5) -> str:
|
|
| 51 |
for it in items[:limit]:
|
| 52 |
mag = _to_float(it.get("magnitudeValue"))
|
| 53 |
depth = _to_float(it.get("depth"))
|
| 54 |
-
tw_str,
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 55 |
lines.append(
|
| 56 |
-
f"事件: {
|
| 57 |
-
f"震級/深度: M{
|
| 58 |
f"時間: {tw_str}(台灣)\n"
|
| 59 |
-
f"預警地區: {
|
| 60 |
)
|
| 61 |
return "\n\n".join(lines).strip()
|
| 62 |
|
|
@@ -105,7 +116,6 @@ def fetch_significant_earthquakes(days: int = 7, limit: int = 5) -> str:
|
|
| 105 |
|
| 106 |
lines = [f"🚨 CWA 最新顯著有感地震 (近{days}天內):", "-" * 20]
|
| 107 |
for _, row in df.iterrows():
|
| 108 |
-
# [修正] 在格式化前,先檢查值是否存在
|
| 109 |
mag_str = f"{row['Magnitude']:.1f}" if pd.notna(row['Magnitude']) else "—"
|
| 110 |
depth_str = f"{row['Depth']:.0f}" if pd.notna(row['Depth']) else "—"
|
| 111 |
|
|
@@ -119,4 +129,3 @@ def fetch_significant_earthquakes(days: int = 7, limit: int = 5) -> str:
|
|
| 119 |
|
| 120 |
except Exception as e:
|
| 121 |
return f"❌ 顯著地震查詢失敗:{e}"
|
| 122 |
-
|
|
|
|
| 28 |
|
| 29 |
# --- 地震預警 (CWA_ALARM_API) ---
|
| 30 |
def fetch_cwa_alarm_list(limit: int = 5) -> str:
|
| 31 |
+
"""抓 CWA 地震預警並格式化輸出。"""
|
| 32 |
try:
|
| 33 |
r = requests.get(CWA_ALARM_API, timeout=10)
|
| 34 |
r.raise_for_status()
|
|
|
|
| 52 |
for it in items[:limit]:
|
| 53 |
mag = _to_float(it.get("magnitudeValue"))
|
| 54 |
depth = _to_float(it.get("depth"))
|
| 55 |
+
tw_str, _ = _parse_cwa_time(it.get("originTime", ""))
|
| 56 |
+
|
| 57 |
+
# [修正] 在格式化前,先對來自 API 的字串進行轉義,避免其中的 { } 字元造成錯誤
|
| 58 |
+
identifier = str(it.get('identifier', '—')).replace('{', '{{').replace('}', '}}')
|
| 59 |
+
msg_type = str(it.get('msgType', '—')).replace('{', '{{').replace('}', '}}')
|
| 60 |
+
msg_no = str(it.get('msgNo', '—')).replace('{', '{{').replace('}', '}}')
|
| 61 |
+
areas = str(it.get('alertAreas') or '—').replace('{', '{{').replace('}', '}}')
|
| 62 |
+
|
| 63 |
+
mag_str = f"{mag:.1f}" if mag is not None else "—"
|
| 64 |
+
depth_str = f"{depth:.0f}" if depth is not None else "—"
|
| 65 |
+
|
| 66 |
lines.append(
|
| 67 |
+
f"事件: {identifier} | 類型: {msg_type}#{msg_no}\n"
|
| 68 |
+
f"震級/深度: M{mag_str} / {depth_str} km\n"
|
| 69 |
f"時間: {tw_str}(台灣)\n"
|
| 70 |
+
f"預警地區: {areas}"
|
| 71 |
)
|
| 72 |
return "\n\n".join(lines).strip()
|
| 73 |
|
|
|
|
| 116 |
|
| 117 |
lines = [f"🚨 CWA 最新顯著有感地震 (近{days}天內):", "-" * 20]
|
| 118 |
for _, row in df.iterrows():
|
|
|
|
| 119 |
mag_str = f"{row['Magnitude']:.1f}" if pd.notna(row['Magnitude']) else "—"
|
| 120 |
depth_str = f"{row['Depth']:.0f}" if pd.notna(row['Depth']) else "—"
|
| 121 |
|
|
|
|
| 129 |
|
| 130 |
except Exception as e:
|
| 131 |
return f"❌ 顯著地震查詢失敗:{e}"
|
|
|