一度 notebook を止めて再起動し
以下のコードを実行した
結果は以下の通り
import requests, feedparser, xml.etree.ElementTree as ET
EQVOL_FEED = "https://www.data.jma.go.jp/developer/xml/feed/eqvol.xml"
NAMESP = {
"jmx": "http://xml.kishou.go.jp/jmaxml1/",
"jmx_ib": "http://xml.kishou.go.jp/jmaxml1/informationBasis1/",
"jmx_eb": "http://xml.kishou.go.jp/jmaxml1/elementBasis1/",
"jmx_seis": "http://xml.kishou.go.jp/jmaxml1/body/seismology1/",
"jmx_tsunami": "http://xml.kishou.go.jp/jmaxml1/body/tsunami1/",
}
def parse_entry_xml(url):
r = requests.get(url, timeout=15)
r.raise_for_status()
root = ET.fromstring(r.content)
head = root.find(".//jmx_ib:Head", NAMESP)
title = head.findtext("jmx_ib:Title", default="", namespaces=NAMESP)
report_dt = head.findtext("jmx_ib:ReportDateTime", default="", namespaces=NAMESP)
info_kind = head.findtext("jmx_ib:InfoKind", default="", namespaces=NAMESP)
info_type = head.findtext("jmx_ib:InfoType", default="", namespaces=NAMESP)
headline = head.findtext("jmx_ib:Headline/jmx_eb:Text", default="", namespaces=NAMESP)
# 津波領域の抽出
areas, kinds = [], []
for item in root.findall(".//jmx_eb:Body//jmx_eb:Information[@type='津波予報領域表現']/jmx_eb:Item", NAMESP):
kind_name = item.findtext("jmx_eb:Kind/jmx_eb:Name", default="", namespaces=NAMESP)
if kind_name:
kinds.append(kind_name)
for a in item.findall("jmx_eb:Areas/jmx_eb:Area", NAMESP):
areas.append({
"name": a.findtext("jmx_eb:Name", default="", namespaces=NAMESP),
"code": a.findtext("jmx_eb:Code", default="", namespaces=NAMESP),
})
return {
"title": title, "report_dt": report_dt,
"info_kind": info_kind, "info_type": info_type,
"headline": headline, "kinds": list(set(kinds)),
"areas": areas, "url": url
}
def is_tsunami_trigger(rec):
# “津波”を広めに判定（タイトル・見出し・InfoKind）
text_pool = " ".join([rec.get("title",""), rec.get("headline",""), rec.get("info_kind","")])
if "津波" not in text_pool:
return False
# 強ワード
kw = ["大津波警報","津波警報","津波注意報","津波予報","津波情報"]
if any(k in text_pool for k in kw):
return True
# あるいはKindsに入っていればOK
return any(k in rec.get("kinds", []) for k in kw)
# ---- ドライラン：feed取得→最新5件だけ詳細をなめて出力 ----
r = requests.get(EQVOL_FEED, timeout=15)
print("feed status:", r.status_code, "bytes:", len(r.content))
d = feedparser.parse(r.text)
print("entries:", len(d.entries))
for e in d.entries[:5]:
print("ENTRY:", e.get("published",""), e.get("title",""))
try:
rec = parse_entry_xml(e.link)
print(" InfoKind:", rec["info_kind"], "| InfoType:", rec["info_type"])
print(" Title :", rec["title"])
print(" Headline:", (rec["headline"] or "")[:120])
print(" kinds :", rec["kinds"], "| areas:", len(rec["areas"]))
print(" -> TRIGGER?", is_tsunami_trigger(rec))
except Exception as ex:
print(" ERROR parsing:", ex)
を実行
feed status: 200 bytes: 27328 entries: 37 ENTRY: 降灰予報（定時） InfoKind: 降灰予報 | InfoType: 発表 Title : 火山名 薩摩硫黄島 降灰予報（定時） Headline: kinds : [] | areas: 0 -> TRIGGER? False ENTRY: 降灰予報（定時） InfoKind: 降灰予報 | InfoType: 発表 Title : 火山名 諏訪之瀬島 降灰予報（定時） Headline: kinds : [] | areas: 0 -> TRIGGER? False ENTRY: 降灰予報（定時） InfoKind: 降灰予報 | InfoType: 発表 Title : 火山名 霧島山（新燃岳） 降灰予報（定時） Headline: kinds : [] | areas: 0 -> TRIGGER? False ENTRY: 降灰予報（定時） InfoKind: 降灰予報 | InfoType: 発表 Title : 火山名 浅間山 降灰予報（定時） Headline: kinds : [] | areas: 0 -> TRIGGER? False ENTRY: 降灰予報（定時） InfoKind: 降灰予報 | InfoType: 発表 Title : 火山名 草津白根山（白根山（湯釜付近）） 降灰予報（定時） Headline: kinds : [] | areas: 0 -> TRIGGER? False
もしこれが
entries=0 → フィード取得が怪しい（ネットワーク・一時ブロック）。requests.get(EQVOL_FEED).text[:2000]をprintして中身を確認
entries>0かつInfoKindやTitleに「津波」が無い → 正常（いま津波電文が無いだけ）
parse_entry_xmlでエラー → XML名前空間/パスが電文種別で違う可能性。print(rec[“url”])で該当XMLを保存して中身を見るのが早い
というように原因を理解できる
補足（実運用へ進めるとき）
静岡沿岸の津波予報区コードでareasをフィルタ（対象外は通知しない）
InfoType == “取消”で状態を解除に遷移
併せてextra.xmlも監視すると取りこぼしが減ります（仕様差し込みは後でOK）
ノートブックではなく常駐スクリプト／systemdで運用し、ログをローテーション
他にやったことは
1. 通信・ネットワークの確認
まず、最初の requests.get(EQVOL_FEED) で通信が成功しているかを確かめる
import requests
EQVOL_FEED = "https://www.data.jma.go.jp/developer/xml/feed/eqvol.xml"
r = requests.get(EQVOL_FEED, timeout=15)
print("status:", r.status_code)
print(r.text[:1000]) # 最初の1000文字だけ表示
✅ status: 200 なら通信成功。
❌ もし 403 や Timeout が出た場合 → 通信がブロックされている（VPN・プロキシ・ファイアウォールなどの影響）。
結果は
status: 200
<?xml version="1.0" encoding="utf-8"?>
<feed xmlns="http://www.w3.org/2005/Atom" lang="ja">
<title>é«é »åº¦ï¼å°éç«å±±ï¼</title>
<subtitle>JMAXML publishing feed</subtitle>
<updated>2025-11-06T05:01:14+09:00</updated>
<id>https://www.data.jma.go.jp/developer/xml/feed/eqvol.xml#short_1762372874</id>
<link rel="related" href="https://www.jma.go.jp/"/>
<link rel="self" href="https://www.data.jma.go.jp/developer/xml/feed/eqvol.xml"/>
<link rel="hub" href="http://alert-hub.appspot.com/"/>
<rights type="html"><![CDATA[
<a href="https://www.jma.go.jp/jma/kishou/info/coment.html">å©ç¨è¦ç´</a>,
<a href="https://www.jma.go.jp/jma/en/copyright.html">Terms of Use</a>
]]></rights>
<entry>
<title>éç°äºå ±ï¼å®æï¼</title>
<id>https://www.data.jma.go.jp/developer/xml/data/20251105200049_0_VFVO53_010000.xml</id>
<updated>2025-11-05T20:00:00Z</updated>
<author>
<name>æ°è±¡åº</name>
</author>
<link type="application/xml" href="https://ww
カーネル・出力の健全性チェック
これで何も出ないなら、セルが実行されていない／カーネルが固まっている／出力パネルが壊れている可能性が濃厚
import sys, time, os, pathlib, subprocess
print("STEP A1: print works?", flush=True)
sys.stdout.write("STEP A2: stdout.write works?\n"); sys.stdout.flush()
sys.stderr.write("STEP A3: stderr works?\n"); sys.stderr.flush()
for i in range(3):
print(f"tick {i}", flush=True)
time.sleep(0.5)
# ファイル出力で実行確認（UI出力が壊れていても痕跡が残る）
pathlib.Path("notebook_alive.txt").write_text("alive\n")
print("STEP A4: wrote notebook_alive.txt", flush=True)
# サブプロセスの標準出力も確認
ret = subprocess.run(["echo", "STEP A5: subprocess echo"], capture_output=True, text=True)
print(ret.stdout, end="")
print("STEP A DONE", flush=True)
結果は
STEP A1: print works? STEP A2: stdout.write works? STEP A3: stderr works? tick 0 tick 1 tick 2 STEP A4: wrote notebook_alive.txt STEP A5: subprocess echo STEP A DONE
しかし
再度
# 津波情報の取得
import time
import requests
import feedparser
import xml.etree.ElementTree as ET
EQVOL_FEED = "https://www.data.jma.go.jp/developer/xml/feed/eqvol.xml"
NAMESP = {
"jmx": "http://xml.kishou.go.jp/jmaxml1/",
"jmx_ib": "http://xml.kishou.go.jp/jmaxml1/informationBasis1/",
"jmx_eb": "http://xml.kishou.go.jp/jmaxml1/elementBasis1/",
"jmx_seis": "http://xml.kishou.go.jp/jmaxml1/body/seismology1/",
"jmx_tsunami": "http://xml.kishou.go.jp/jmaxml1/body/tsunami1/",
}
last_modified = None
seen_ids = set()
def fetch_feed():
headers = {}
if last_modified:
headers["If-Modified-Since"] = last_modified
r = requests.get(EQVOL_FEED, headers=headers, timeout=10)
if r.status_code == 304:
return None, None
r.raise_for_status()
return r.text, r.headers.get("Last-Modified")
def parse_entry_xml(url):
r = requests.get(url, timeout=10)
r.raise_for_status()
root = ET.fromstring(r.content)
head = root.find(".//jmx_ib:Head", NAMESP)
title = head.findtext("jmx_ib:Title", default="", namespaces=NAMESP)
report_dt = head.findtext("jmx_ib:ReportDateTime", default="", namespaces=NAMESP)
info_kind = head.findtext("jmx_ib:InfoKind", default="", namespaces=NAMESP)
info_type = head.findtext("jmx_ib:InfoType", default="", namespaces=NAMESP)
headline = head.findtext("jmx_ib:Headline/jmx_eb:Text", default="", namespaces=NAMESP)
# 津波関連の抽出（領域と種別）
areas, kinds = [], []
for item in root.findall(".//jmx_eb:Body//jmx_eb:Information[@type='津波予報領域表現']/jmx_eb:Item", NAMESP):
kind_name = item.findtext("jmx_eb:Kind/jmx_eb:Name", default="", namespaces=NAMESP)
kinds.append(kind_name)
for a in item.findall("jmx_eb:Areas/jmx_eb:Area", NAMESP):
areas.append({
"name": a.findtext("jmx_eb:Name", default="", namespaces=NAMESP),
"code": a.findtext("jmx_eb:Code", default="", namespaces=NAMESP),
})
return {
"title": title,
"report_dt": report_dt,
"info_kind": info_kind,
"info_type": info_type,
"headline": headline,
"kinds": list(set(kinds)),
"areas": areas,
"url": url,
}
def is_tsunami_trigger(rec):
# 種別に“津波警報・注意報・予報/津波情報”などが含まれるものを対象
if "津波" not in (rec["info_kind"] or rec["title"] or rec["headline"]):
return False
# 警報/注意報/予報の明示があれば強トリガー
keywords = ["大津波警報", "津波警報", "津波注意報", "津波予報", "津波情報"]
if any(k in (rec["headline"] or "") for k in keywords):
return True
# 後方互換（kindsに入っていればOK）
return any(k in keywords for k in rec["kinds"])
def loop():
global last_modified
while True:
try:
feed_text, lm = fetch_feed()
if feed_text:
d = feedparser.parse(feed_text)
# 新しい順に処理
for e in d.entries:
if e.id in seen_ids:
continue
seen_ids.add(e.id)
rec = parse_entry_xml(e.link)
if is_tsunami_trigger(rec):
# === ここであなたの通知/保存パイプラインへ ===
print("[TSUNAMI TRIGGER]", rec["report_dt"], rec["headline"], rec["kinds"][:3], "areas=", len(rec["areas"]))
if lm:
last_modified = lm
except Exception as ex:
print("error:", ex)
time.sleep(60)
if __name__ == "__main__":
loop()
を実行すると終わらない
ノートブックで「終わらない」のは、**無限ループ中に“トリガー時しかprintしない”**から
＝ 津波電文が無い通常時は1分ごとに静かにsleep→再取得、を永遠に繰り返す
すぐ直すポイント（Notebook向け）
1. デバッグ出力を入れて「今なにしてるか」を毎ループ表示（flushも）
2. ループ回数に上限（MAX_LOOPS）を付けてセルが必ず終わる
3. 取得件数・新規処理件数・直近タイトルなどを表示
4. 例外時にURLも表示して切り分けしやすく
その前に処理を停止する方法を調べる
すぐ止める（今まさに動いているセルを止める）
* JupyterLab：Esc → I を素早く2回（= Interrupt）。またはツールバーの■（Stop）をクリック
* Classic Notebook：メニュー Kernel → Interrupt
* Google Colab：メニュー ランタイム → 実行を中断（ショートカット：Ctrl/Cmd + M, I）
* それでも止まらない/固まる時：Kernel → Restart（再起動）
これで KeyboardInterrupt が発生し、セルの実行が中断されます。
ということで
🔳アイコンをクリックして処理を停止
---------------------------------------------------------------------------
KeyboardInterrupt Traceback (most recent call last)
Cell In[4], line 97
94 time.sleep(60)
96 if __name__ == "__main__":
---> 97 loop()
Cell In[4], line 94, in loop()
92 except Exception as ex:
93 print("error:", ex)
---> 94 time.sleep(60)
KeyboardInterrupt:
と表示された
停止確認できたので
デバッグ出力を入れて「今なにしてるか」を毎ループ表示（flushも）
ループ回数に上限（MAX_LOOPS）を付けてセルが必ず終わる
取得件数・新規処理件数・直近タイトルなどを表示
例外時にURLも表示して切り分けしやすく
を入れる
Notebook向けに
* 3周で必ず終了（MAX_LOOPS=3）
* 毎ループの状況を表示（entries / new / last_modified と直近タイトル）
* 変更なしのときは 304 Not Modified と出力
になるので、「動いているのに何も出ない」を防げる
小さな実用メモ
* 早く挙動を見たいときは SLEEP_SEC = 15 などにするとテンポ良く確認できます。
* 何度かセルを繰り返し実行するなら、毎回まっさらで試すために先頭で seen_ids = set() を明示的に初期化しておくと安心です。
* 津波電文が出たときだけ [TSUNAMI TRIGGER] … が出ます（通常時は出ません）。
次の一手（お好みで）
* 対象地域フィルタ：静岡沿岸などの予報区コードだけ通知したい場合は、parse_entry_xml() で得た rec[“areas”] を使って絞れます（例：if not any(a[“code”] in SHIZUOKA_CODES for a in rec[“areas”]): return False）。コードは後で差し込みましょう（リストは用意します）。
* 見栄えの良い停止：loop() を try: … except KeyboardInterrupt: print(“Stopped by user”) で包むとトレース無しで止まります。
* 常駐運用：Notebook検証が終わったら MAX_LOOPS を外し、systemd などで常駐＋ログローテーションにすると本番向きです。
ということで
# Notebookデモ用：必ず終わる・状況が見える版
import time, requests, feedparser, xml.etree.ElementTree as ET
EQVOL_FEED = "https://www.data.jma.go.jp/developer/xml/feed/eqvol.xml"
NAMESP = {
"jmx": "http://xml.kishou.go.jp/jmaxml1/",
"jmx_ib": "http://xml.kishou.go.jp/jmaxml1/informationBasis1/",
"jmx_eb": "http://xml.kishou.go.jp/jmaxml1/elementBasis1/",
"jmx_seis": "http://xml.kishou.go.jp/jmaxml1/body/seismology1/",
"jmx_tsunami": "http://xml.kishou.go.jp/jmaxml1/body/tsunami1/",
}
last_modified = None
seen_ids = set()
MAX_LOOPS = 3 # ★ Notebookでは必ず終わらせる
SLEEP_SEC = 60 # デモなら 15 にしてもOK
DEBUG = True
def fetch_feed():
headers = {}
if last_modified:
headers["If-Modified-Since"] = last_modified
r = requests.get(EQVOL_FEED, headers=headers, timeout=10)
if r.status_code == 304:
return None, None
r.raise_for_status()
return r.text, r.headers.get("Last-Modified")
def parse_entry_xml(url):
r = requests.get(url, timeout=10)
r.raise_for_status()
root = ET.fromstring(r.content)
head = root.find(".//jmx_ib:Head", NAMESP)
title = head.findtext("jmx_ib:Title", default="", namespaces=NAMESP)
report_dt = head.findtext("jmx_ib:ReportDateTime", default="", namespaces=NAMESP)
info_kind = head.findtext("jmx_ib:InfoKind", default="", namespaces=NAMESP)
info_type = head.findtext("jmx_ib:InfoType", default="", namespaces=NAMESP)
headline = head.findtext("jmx_ib:Headline/jmx_eb:Text", default="", namespaces=NAMESP)
areas, kinds = [], []
for item in root.findall(".//jmx_eb:Body//jmx_eb:Information[@type='津波予報領域表現']/jmx_eb:Item", NAMESP):
kind_name = item.findtext("jmx_eb:Kind/jmx_eb:Name", default="", namespaces=NAMESP)
if kind_name:
kinds.append(kind_name)
for a in item.findall("jmx_eb:Areas/jmx_eb:Area", NAMESP):
areas.append({
"name": a.findtext("jmx_eb:Name", default="", namespaces=NAMESP),
"code": a.findtext("jmx_eb:Code", default="", namespaces=NAMESP),
})
return {
"title": title, "report_dt": report_dt, "info_kind": info_kind, "info_type": info_type,
"headline": headline, "kinds": list(set(kinds)), "areas": areas, "url": url
}
def is_tsunami_trigger(rec):
pool = " ".join([rec.get("title",""), rec.get("headline",""), rec.get("info_kind","")])
if "津波" not in pool:
return False
kw = ["大津波警報","津波警報","津波注意報","津波予報","津波情報"]
if any(k in pool for k in kw):
return True
return any(k in rec.get("kinds", []) for k in kw)
def loop():
global last_modified
loops = 0
while True:
try:
feed_text, lm = fetch_feed()
if feed_text:
d = feedparser.parse(feed_text)
new_cnt = 0
last_titles = []
for e in d.entries:
if e.id in seen_ids:
continue
seen_ids.add(e.id)
new_cnt += 1
last_titles.append(e.title)
rec = parse_entry_xml(e.link)
if is_tsunami_trigger(rec):
print("[TSUNAMI TRIGGER]", rec["report_dt"], rec["headline"], rec["kinds"][:3], "areas=", len(rec["areas"]), flush=True)
if DEBUG:
print(f"[{time.strftime('%H:%M:%S')}] entries={len(d.entries)} new={new_cnt} last_modified={lm}", flush=True)
if last_titles:
print(" recent:", " / ".join(last_titles[:3])[:160], flush=True)
else:
print(" recent: (no new entries this cycle)", flush=True)
else:
if DEBUG:
print(f"[{time.strftime('%H:%M:%S')}] 304 Not Modified（変更なし）", flush=True)
if lm:
last_modified = lm
except Exception as ex:
print("error:", ex, flush=True)
loops += 1
if loops >= MAX_LOOPS:
print("done (MAX_LOOPS reached)", flush=True)
break
time.sleep(SLEEP_SEC)
loop()
に変更して実行する
ログは「フィード取得は成功・“いま津波電文はない”・初回なので new=33（初回は未既知IDがないため全部新規扱い）」という正常動作
最近流れているのが降灰予報（火山）なので、[TSUNAMI TRIGGER] は出ない
無駄なXML取得を減らして津波だけを見る・静岡沿岸だけ通知・既読IDを永続化の3点を足す
# 津波専用・Notebook検証向け（津波以外はスキップ、静岡コードで絞り、既読IDを永続化）
import time, json, os
import requests, feedparser, xml.etree.ElementTree as ET
EQVOL_FEED = "https://www.data.jma.go.jp/developer/xml/feed/eqvol.xml"
NAMESP = {
"jmx": "http://xml.kishou.go.jp/jmaxml1/",
"jmx_ib": "http://xml.kishou.go.jp/jmaxml1/informationBasis1/",
"jmx_eb": "http://xml.kishou.go.jp/jmaxml1/elementBasis1/",
"jmx_seis": "http://xml.kishou.go.jp/jmaxml1/body/seismology1/",
"jmx_tsunami": "http://xml.kishou.go.jp/jmaxml1/body/tsunami1/",
}
# --- 静岡沿岸の津波予報区コード（代表例）：必要に応じて追加 ---
SHIZUOKA_CODES = {
"504", # 遠州灘
"505", # 駿河湾
# 必要なら追加（JMAコード表に準拠）
}
last_modified = None
SEEN_FILE = "seen_ids.json"
seen_ids = set(json.load(open(SEEN_FILE)) if os.path.exists(SEEN_FILE) else [])
MAX_LOOPS = 3
SLEEP_SEC = 60
DEBUG = True
def save_seen():
try:
with open(SEEN_FILE, "w") as f:
json.dump(list(seen_ids), f)
except Exception:
pass
def fetch_feed():
headers = {}
if last_modified:
headers["If-Modified-Since"] = last_modified
r = requests.get(EQVOL_FEED, headers=headers, timeout=10)
if r.status_code == 304:
return None, None
r.raise_for_status()
return r.text, r.headers.get("Last-Modified")
def parse_entry_xml(url):
r = requests.get(url, timeout=10)
r.raise_for_status()
root = ET.fromstring(r.content)
head = root.find(".//jmx_ib:Head", NAMESP)
rec = {
"title": head.findtext("jmx_ib:Title", default="", namespaces=NAMESP),
"report_dt": head.findtext("jmx_ib:ReportDateTime", default="", namespaces=NAMESP),
"info_kind": head.findtext("jmx_ib:InfoKind", default="", namespaces=NAMESP),
"info_type": head.findtext("jmx_ib:InfoType", default="", namespaces=NAMESP),
"headline": head.findtext("jmx_ib:Headline/jmx_eb:Text", default="", namespaces=NAMESP),
"kinds": [],
"areas": [],
"url": url,
}
# 津波領域（情報タイプ=津波予報領域表現）
for item in root.findall(".//jmx_eb:Body//jmx_eb:Information[@type='津波予報領域表現']/jmx_eb:Item", NAMESP):
k = item.findtext("jmx_eb:Kind/jmx_eb:Name", default="", namespaces=NAMESP)
if k: rec["kinds"].append(k)
for a in item.findall("jmx_eb:Areas/jmx_eb:Area", NAMESP):
rec["areas"].append({
"name": a.findtext("jmx_eb:Name", default="", namespaces=NAMESP),
"code": a.findtext("jmx_eb:Code", default="", namespaces=NAMESP),
})
rec["kinds"] = list(set(rec["kinds"]))
return rec
def is_tsunami_trigger(rec):
pool = " ".join([rec.get("title",""), rec.get("headline",""), rec.get("info_kind","")])
if "津波" not in pool:
return False
kw = ["大津波警報","津波警報","津波注意報","津波予報","津波情報"]
if any(k in pool for k in kw):
return True
return any(k in rec.get("kinds", []) for k in kw)
def is_target_area(rec):
if not rec["areas"]:
# 領域表現が無い電文は地域未特定なので一旦通す or 無視するかは方針次第
return True
return any(a.get("code") in SHIZUOKA_CODES for a in rec["areas"])
def loop():
global last_modified
loops = 0
while True:
try:
feed_text, lm = fetch_feed()
if feed_text:
d = feedparser.parse(feed_text)
# ★ フィード段階で「津波」を含むエントリだけに早期フィルタ
tsunami_entries = [e for e in d.entries
if "津波" in (e.title or "") or "tsunami" in (e.title or "").lower()]
new_cnt = 0
for e in tsunami_entries:
if e.id in seen_ids:
continue
seen_ids.add(e.id); new_cnt += 1
rec = parse_entry_xml(e.link)
if is_tsunami_trigger(rec) and is_target_area(rec):
print("[TSUNAMI TRIGGER]", rec["report_dt"], rec["headline"],
rec["kinds"][:3], "areas=", len(rec["areas"]), flush=True)
if DEBUG:
print(f"[{time.strftime('%H:%M:%S')}] entries={len(d.entries)} tsunami_only={len(tsunami_entries)} new={new_cnt} last_modified={lm}", flush=True)
save_seen()
else:
if DEBUG:
print(f"[{time.strftime('%H:%M:%S')}] 304 Not Modified（変更なし）", flush=True)
if lm:
last_modified = lm
except Exception as ex:
print("error:", ex, flush=True)
loops += 1
if loops >= MAX_LOOPS:
print("done (MAX_LOOPS reached)", flush=True)
break
time.sleep(SLEEP_SEC)
loop()
で実行
[06:25:13] entries=33 tsunami_only=0 new=0 last_modified=Fri, 07 Nov 2025 20:01:38 GMT [06:26:13] 304 Not Modified（変更なし） [06:27:14] 304 Not Modified（変更なし） done (MAX_LOOPS reached)
となる
tsunami_only=0 は「今回のフィードのエントリータイトルに“津波”が一件も無い」という意味で、現状は降灰や地震系だけ＝津波電文が出ていない状態です。
ただし——タイトルに“津波”が無くても本体XML側が津波情報というケースがあるため、実運用では「タイトルで早期フィルタ」は外して、直近N件だけ中身を開いて判定するのが堅い
安全版（タイトルで絞らず、直近N件だけ本文チェック／extra.xml併走）
直近 MAX_FETCH=10 件だけ本文XMLを開いて InfoKind/Headline から津波判定
eqvol.xml と extra.xml を同じロジックで監視
既読IDはフィードごとに保存
# 津波検知（タイトルで絞らず本文チェック）+ extra.xml 併走 / Notebook用
import time, json, os, requests, feedparser, xml.etree.ElementTree as ET
FEEDS = {
"eqvol": "https://www.data.jma.go.jp/developer/xml/feed/eqvol.xml",
"extra": "https://www.data.jma.go.jp/developer/xml/feed/extra.xml",
}
NAMESP = {
"jmx": "http://xml.kishou.go.jp/jmaxml1/",
"jmx_ib": "http://xml.kishou.go.jp/jmaxml1/informationBasis1/",
"jmx_eb": "http://xml.kishou.go.jp/jmaxml1/elementBasis1/",
"jmx_seis": "http://xml.kishou.go.jp/jmaxml1/body/seismology1/",
"jmx_tsunami": "http://xml.kishou.go.jp/jmaxml1/body/tsunami1/",
}
# 静岡の津波予報区コード（例）
SHIZUOKA_CODES = {"504", "505"} # 遠州灘/駿河湾。必要なら追加
MAX_FETCH = 10 # 各サイクルで本文を開く最大件数
MAX_LOOPS = 3 # Notebook用：必ず終わる
SLEEP_SEC = 60
DEBUG = True
state = {name: {"last_modified": None, "seen": set()} for name in FEEDS}
def _seen_path(feed_name): return f"seen_ids_{feed_name}.json"
# 既読IDのロード
for name in FEEDS:
p = _seen_path(name)
if os.path.exists(p):
try:
state[name]["seen"] = set(json.load(open(p)))
except Exception:
pass
def save_seen(feed_name):
try:
json.dump(list(state[feed_name]["seen"]), open(_seen_path(feed_name), "w"))
except Exception:
pass
def fetch_feed(feed_name, url):
headers = {}
lm = state[feed_name]["last_modified"]
if lm: headers["If-Modified-Since"] = lm
r = requests.get(url, headers=headers, timeout=10)
if r.status_code == 304:
return None, None
r.raise_for_status()
return r.text, r.headers.get("Last-Modified")
def parse_entry_xml(url):
r = requests.get(url, timeout=10)
r.raise_for_status()
root = ET.fromstring(r.content)
head = root.find(".//jmx_ib:Head", NAMESP)
rec = {
"title": head.findtext("jmx_ib:Title", default="", namespaces=NAMESP),
"report_dt": head.findtext("jmx_ib:ReportDateTime", default="", namespaces=NAMESP),
"info_kind": head.findtext("jmx_ib:InfoKind", default="", namespaces=NAMESP),
"info_type": head.findtext("jmx_ib:InfoType", default="", namespaces=NAMESP),
"headline": head.findtext("jmx_ib:Headline/jmx_eb:Text", default="", namespaces=NAMESP),
"kinds": [], "areas": [], "url": url,
}
# 津波予報領域
for item in root.findall(".//jmx_eb:Body//jmx_eb:Information[@type='津波予報領域表現']/jmx_eb:Item", NAMESP):
k = item.findtext("jmx_eb:Kind/jmx_eb:Name", default="", namespaces=NAMESP)
if k: rec["kinds"].append(k)
for a in item.findall("jmx_eb:Areas/jmx_eb:Area", NAMESP):
rec["areas"].append({
"name": a.findtext("jmx_eb:Name", default="", namespaces=NAMESP),
"code": a.findtext("jmx_eb:Code", default="", namespaces=NAMESP),
})
rec["kinds"] = list(set(rec["kinds"]))
return rec
def is_tsunami_trigger(rec):
pool = " ".join([rec.get("title",""), rec.get("headline",""), rec.get("info_kind","")])
if "津波" not in pool: # 見出し系に「津波」が登場しないものは通常スキップ
return False
kw = ["大津波警報","津波警報","津波注意報","津波予報","津波情報"]
if any(k in pool for k in kw): return True
return any(k in rec.get("kinds", []) for k in kw)
def is_target_area(rec):
# 領域未記載なら方針次第。ここでは一旦通す。
if not rec["areas"]: return True
return any(a.get("code") in SHIZUOKA_CODES for a in rec["areas"])
def watch_cycle():
total_new = 0
for name, url in FEEDS.items():
try:
feed_text, lm = fetch_feed(name, url)
if not feed_text:
if DEBUG: print(f"[{name}] 304 Not Modified")
continue
d = feedparser.parse(feed_text)
# まだ見てない最新順から MAX_FETCH 件だけ本文チェック
new_entries = [e for e in d.entries if e.id not in state[name]["seen"]][:MAX_FETCH]
total_new += len(new_entries)
for e in new_entries:
state[name]["seen"].add(e.id)
rec = parse_entry_xml(e.link)
if is_tsunami_trigger(rec) and is_target_area(rec):
print("[TSUNAMI TRIGGER]", name, rec["report_dt"], rec["headline"],
rec["kinds"][:3], "areas=", len(rec["areas"]), flush=True)
if DEBUG:
print(f"[{name}] entries={len(d.entries)} checked={len(new_entries)} last_modified={lm}")
save_seen(name)
if lm: state[name]["last_modified"] = lm
except Exception as ex:
print(f"[{name}] error:", ex, flush=True)
return total_new
# ===== Notebook用メインループ =====
loops = 0
while loops < MAX_LOOPS:
_ = watch_cycle()
loops += 1
if loops >= MAX_LOOPS:
print("done (MAX_LOOPS reached)")
break
time.sleep(SLEEP_SEC)
で実行
タイトルに“津波”が無いせいで tsunami_only=0 になる問題を回避（本文で判定）
フィード2本（eqvol/extra）を同ロジックで監視
既読の永続化により、Notebookを再実行しても初回から「全部new」にならない
直近 MAX_FETCH のみ本文取得で無駄アクセス抑制
結果は
[eqvol] entries=33 checked=10 last_modified=Fri, 07 Nov 2025 20:01:38 GMT [extra] entries=106 checked=10 last_modified=Fri, 07 Nov 2025 21:24:15 GMT [eqvol] 304 Not Modified [extra] 304 Not Modified
ログの意味
* [eqvol] entries=33 checked=10：フィードに33件、今回は未読の先頭10件だけ本文を精査
* [extra] entries=106 checked=10：同上
* 304 Not Modified：前回から更新なし＝正常
いまは津波電文が無いのでトリガーなし＝想定どおり
とりあえず動作はしそうなので
次は通知の方法を考える