116 lines
3.4 KiB
Python
116 lines
3.4 KiB
Python
import datetime
|
||
from nonebot_plugin_alconna import UniMessage
|
||
from konabot.common.apis.wolfx import CencEewReport, CencEqReport, wolfx_api
|
||
from konabot.common.subscribe import PosterInfo, broadcast, register_poster_info
|
||
|
||
|
||
provinces_short = [
|
||
"北京",
|
||
"天津",
|
||
"河北",
|
||
"山西",
|
||
"内蒙古",
|
||
"辽宁",
|
||
"吉林",
|
||
"黑龙江",
|
||
"上海",
|
||
"江苏",
|
||
"浙江",
|
||
"安徽",
|
||
"福建",
|
||
"江西",
|
||
"山东",
|
||
"河南",
|
||
"湖北",
|
||
"湖南",
|
||
"广东",
|
||
"广西",
|
||
"海南",
|
||
"重庆",
|
||
"四川",
|
||
"贵州",
|
||
"云南",
|
||
"西藏",
|
||
"陕西",
|
||
"甘肃",
|
||
"青海",
|
||
"宁夏",
|
||
"新疆",
|
||
"香港",
|
||
"澳门",
|
||
"台湾",
|
||
]
|
||
|
||
|
||
register_poster_info(
|
||
"中国地震台网地震速报",
|
||
PosterInfo(
|
||
aliases={
|
||
"地震速报",
|
||
"地震预警",
|
||
},
|
||
description="来自中国地震台网的地震速报",
|
||
),
|
||
)
|
||
|
||
|
||
CENC_EEW_DISABLED = True
|
||
|
||
|
||
if not CENC_EEW_DISABLED:
|
||
|
||
@wolfx_api.cenc_eew.append
|
||
async def broadcast_eew(report: CencEewReport):
|
||
# 这个好像没那么准确...
|
||
is_cn = any(report.HypoCenter.startswith(prefix) for prefix in provinces_short)
|
||
if (is_cn and report.Magnitude >= 4.2) or (
|
||
(not is_cn) and report.Magnitude >= 7.0
|
||
):
|
||
# 这是中国地震台网网站上，会默认展示的地震信息的等级
|
||
origin_time_dt = datetime.datetime.strptime(
|
||
report.OriginTime, "%Y-%m-%d %H:%M:%S"
|
||
)
|
||
origin_time_str = (
|
||
f"{origin_time_dt.month}月"
|
||
f"{origin_time_dt.day}日"
|
||
f"{origin_time_dt.hour}时"
|
||
f"{origin_time_dt.minute}分"
|
||
)
|
||
|
||
# vvv 下面这个其实不准确
|
||
eid_in_link = report.EventID.split(".")[0]
|
||
link = f"https://www.cenc.ac.cn/earthquake-manage-publish-web/product-list/{eid_in_link}/summarize"
|
||
|
||
msg = UniMessage.text(
|
||
"据中国地震台网中心 (https://www.cenc.ac.cn/) 报道，"
|
||
f"北京时间{origin_time_str}，"
|
||
f"{report.HypoCenter}发生{report.Magnitude:.1f}级地震。"
|
||
f"震源位于 {report.Longitude}° {report.Latitude}°，深度 {report.Depth}km。\n\n"
|
||
f"详细信息请见 {link}"
|
||
)
|
||
await broadcast("中国地震台网地震速报", msg)
|
||
|
||
|
||
@wolfx_api.cenc_eqlist.append
|
||
async def broadcast_cenc_eqlist(report: CencEqReport):
|
||
is_cn = any(report.location.startswith(prefix) for prefix in provinces_short)
|
||
if (is_cn and float(report.magnitude) >= 4.2) or (
|
||
(not is_cn) and float(report.magnitude) >= 7.0
|
||
):
|
||
origin_time_dt = datetime.datetime.strptime(report.time, "%Y-%m-%d %H:%M:%S")
|
||
origin_time_str = (
|
||
f"{origin_time_dt.month}月"
|
||
f"{origin_time_dt.day}日"
|
||
f"{origin_time_dt.hour}时"
|
||
f"{origin_time_dt.minute}分"
|
||
)
|
||
|
||
msg = UniMessage.text(
|
||
"据中国地震台网中心 (https://www.cenc.ac.cn/) 消息，"
|
||
f"北京时间{origin_time_str}，"
|
||
f"{report.location}发生{report.magnitude}级地震。"
|
||
f"震源位于 {report.longtitude}° {report.latitude}°，深度 {report.depth}km。\n\n"
|
||
f"数据来源于 Wolfx OpenAPI，事件 ID: {report.EventID}"
|
||
)
|
||
await broadcast("中国地震台网地震速报", msg)
|