import requests
from bs4 import BeautifulSoup
import pdfplumber
import io
import json
import time
import sqlite3
import uuid
from urllib.parse import urljoin
from google import genai
from google.genai import types
# ==========================================
# 1. 初期設定
# ==========================================
API_KEY = "AAA" # 実際のキーに置き換えてください
client = genai.Client(api_key=API_KEY)
TARGET_PAGE_URL = "https://www.data.jma.go.jp/sapporo/bosai/past_kishou/kishou.html"
DB_NAME = "weather_disasters.db"
# ==========================================
# 2. データベース操作関数
# ==========================================
def setup_database():
conn = sqlite3.connect(DB_NAME)
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS disaster_events (
event_id TEXT PRIMARY KEY,
start_datetime_utc TEXT,
end_datetime_utc TEXT,
location_mesh TEXT,
location_name TEXT,
weather_pattern TEXT,
hazard_type TEXT,
social_impact TEXT,
alert_level INTEGER,
gpv_archive_path TEXT,
source_url TEXT
)
''')
conn.commit()
conn.close()
def is_already_processed(source_url):
"""URLがすでにDBに存在するかチェックする"""
conn = sqlite3.connect(DB_NAME)
cursor = conn.cursor()
cursor.execute('SELECT 1 FROM disaster_events WHERE source_url LIKE ?', (f"%{source_url}%",))
result = cursor.fetchone()
conn.close()
return result is not None
def find_similar_event(start_datetime_utc):
"""【追加】発生日時(UTC)から、±3日以内の同一事象と思われるレコードを検索する"""
if not start_datetime_utc:
return None
conn = sqlite3.connect(DB_NAME)
conn.row_factory = sqlite3.Row # カラム名でアクセスできるようにする
cursor = conn.cursor()
# SQLiteのjulianday関数を利用して、UTC時間同士の差分が3日以内のものを取得
cursor.execute('''
SELECT * FROM disaster_events
WHERE abs(julianday(start_datetime_utc) - julianday(?)) <= 3
LIMIT 1
''', (start_datetime_utc,))
row = cursor.fetchone()
conn.close()
if row:
return dict(row)
return None
def update_disaster_data(event_id, merged_impact, existing_url, new_url):
"""【追加】既存レコードの被害テキストを更新し、URLを追記する"""
conn = sqlite3.connect(DB_NAME)
cursor = conn.cursor()
# 重複を避けてURLをカンマ区切りで結合
updated_url = existing_url
if new_url not in existing_url:
updated_url = f"{existing_url},{new_url}"
cursor.execute('''
UPDATE disaster_events
SET social_impact = ?, source_url = ?
WHERE event_id = ?
''', (merged_impact, updated_url, event_id))
conn.commit()
conn.close()
print(" -> 既存の事象に情報をマージ（統合）し、データベースを更新しました。")
def insert_disaster_data(data, source_url):
"""新規事象としてデータベースに保存する（変更なし）"""
if isinstance(data, list):
if len(data) > 0:
data = data[0]
else:
print(" -> エラー: AIの出力データが空です。")
return
conn = sqlite3.connect(DB_NAME)
cursor = conn.cursor()
event_id = str(uuid.uuid4())
cursor.execute('''
INSERT INTO disaster_events (
event_id, start_datetime_utc, end_datetime_utc, location_mesh,
location_name, weather_pattern, hazard_type, social_impact,
alert_level, gpv_archive_path, source_url
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
''', (
event_id,
data.get("start_datetime_utc"),
data.get("end_datetime_utc"),
None,
data.get("location_name"),
data.get("weather_pattern"),
data.get("hazard_type"),
data.get("social_impact"),
data.get("alert_level"),
"",
source_url
))
conn.commit()
conn.close()
print(" -> 新規事象としてデータベースへの保存が完了しました。")
# ==========================================
# 3. スクレイピング・AI解析関数
# ==========================================
def get_pdf_links(page_url):
print(f"ページを解析中: {page_url}")
headers = {'User-Agent': 'Mozilla/5.0'}
response = requests.get(page_url, headers=headers)
response.encoding = response.apparent_encoding
soup = BeautifulSoup(response.text, 'html.parser')
pdf_links = []
for a_tag in soup.find_all('a', href=True):
href = a_tag['href']
if href.lower().endswith('.pdf'):
full_url = urljoin(page_url, href)
pdf_links.append(full_url)
return pdf_links
def extract_text_from_pdf_url(pdf_url):
headers = {'User-Agent': 'Mozilla/5.0'}
response = requests.get(pdf_url, headers=headers)
if response.status_code != 200:
return ""
pdf_memory = io.BytesIO(response.content)
extracted_text = ""
try:
with pdfplumber.open(pdf_memory) as pdf:
for page in pdf.pages:
text = page.extract_text()
if text:
extracted_text += text + "\n"
except Exception as e:
pass
return extracted_text
def analyze_disaster_text_with_ai(text):
print("AIでテキストを構造化・UTC変換中...")
prompt = f"""
あなたは気象災害の専門家です。以下の報告書テキストから、災害事例データベース用の情報を抽出してください。
【厳守するルール】
1. テキスト内の日時は日本標準時（JST）です。気象データとの同期エラーを防ぐため、出力するJSON内の日時は「必ずUTC（協定世界時）に変換して」ISO8601形式で出力してください。（例: JSTで3月2日09:00 -> UTCで3月2日00:00）
2. 警戒レベル（alert_level）は、死者や大規模な立ち往生、都市機能停止がある場合は「5」、運休・通行止め等は「4」、遅延は「3」、その他は「2」として判定してください。
【出力JSONスキーマ】
{{
"start_datetime_utc": "YYYY-MM-DDTHH:MM:SSZ",
"end_datetime_utc": "YYYY-MM-DDTHH:MM:SSZ",
"location_name": "発生場所の文字列",
"weather_pattern": "石狩湾小低気圧、JPCZ、発達した低気圧などの気象要因",
"hazard_type": "大雪、暴風雪、浸水などの災害種別",
"social_impact": "交通障害やインフラ被害などの具体的な要約",
"alert_level": 警戒レベルの数値(1-5)
}}
【報告書テキスト】
{text[:5000]}
"""
try:
response = client.models.generate_content(
model='gemini-3.1-flash-lite',
contents=prompt,
config=types.GenerateContentConfig(
response_mime_type="application/json",
),
)
return json.loads(response.text)
except Exception as e:
print(f"AI解析エラー: {e}")
return None
def merge_disaster_text_with_ai(existing_text, new_text):
"""【追加】既存の被害テキストと新しい被害テキストをAIでマージする"""
print(" -> データベース内に類似事象を発見。AIで被害内容を統合中...")
prompt = f"""
あなたは気象災害の専門家です。以下の2つの被害報告は、同時期に起きた同一の気象災害に関する別々の資料からの抜粋です。
重複する内容を整理し、矛盾なく1つの網羅的で詳細な被害テキスト（social_impact）としてまとめ直してください。
出力は統合されたテキスト（文字列）のみとし、JSON等の装飾は不要です。
【既存の被害報告】
{existing_text}
【新しい被害報告】
{new_text}
"""
try:
response = client.models.generate_content(
model='gemini-3.1-flash-lite',
contents=prompt,
)
return response.text.strip()
except Exception as e:
print(f"マージ時のAI解析エラー: {e}")
return f"{existing_text} / {new_text}" # エラー時は安全のため単純結合
# ==========================================
# 4. メイン処理（重複チェック ＆ マージ機能追加）
# ==========================================
if __name__ == "__main__":
setup_database()
pdf_urls = get_pdf_links(TARGET_PAGE_URL)
print(f"見つかったPDFの数: {len(pdf_urls)}件\n")
for i, pdf_url in enumerate(pdf_urls):
print(f"\n--- {i+1}件目の処理を開始 ({pdf_url}) ---")
if is_already_processed(pdf_url):
print(" -> このURLは既にデータベースに登録されているためスキップします。")
continue
raw_text = extract_text_from_pdf_url(pdf_url)
if raw_text:
max_retries = 3
for attempt in range(max_retries):
structured_data = analyze_disaster_text_with_ai(raw_text)
if structured_data:
print("【AI解析成功】")
# リスト型対策（辞書型を取り出す）
data_dict = structured_data
if isinstance(data_dict, list):
data_dict = data_dict[0] if len(data_dict) > 0 else {}
# 1. 時間軸をベースに類似の既存事象を検索
start_dt = data_dict.get("start_datetime_utc")
similar_event = find_similar_event(start_dt)
if similar_event:
# 2. 類似事象があれば、被害内容(social_impact)をAIでマージ
existing_impact = similar_event.get("social_impact", "")
new_impact = data_dict.get("social_impact", "")
merged_impact = merge_disaster_text_with_ai(existing_impact, new_impact)
# 3. 既存レコードをUPDATE
update_disaster_data(similar_event["event_id"], merged_impact, similar_event["source_url"], pdf_url)
else:
# 4. 類似事象がなければ新規INSERT
insert_disaster_data(data_dict, pdf_url)
break
else:
print(f"解析失敗。30秒待機して再試行します... (試行 {attempt + 1}/{max_retries})")
time.sleep(30)
time.sleep(10)
def extract_text_from_html_url(html_url):
"""Webページ（HTML）からノイズを除去して本文テキストを抽出する"""
print(f"HTMLページをダウンロード・抽出中: {html_url}")
headers = {'User-Agent': 'Mozilla/5.0'}
try:
response = requests.get(html_url, headers=headers, timeout=10)
response.encoding = response.apparent_encoding # 文字化け防止
if response.status_code != 200:
print(f"HTMLの取得に失敗しました: {response.status_code}")
return ""
soup = BeautifulSoup(response.text, 'html.parser')
# AIの解析ノイズになる不要なタグ（スクリプト、スタイル、ヘッダー等）を削除
for script_or_style in soup(['script', 'style', 'header', 'footer', 'nav']):
script_or_style.decompose()
# 本文テキストを抽出し、余分な改行や空白を整理する
text = soup.get_text(separator='\n')
lines = (line.strip() for line in text.splitlines())
chunks = (phrase.strip() for line in lines for phrase in line.split(" "))
extracted_text = '\n'.join(chunk for chunk in chunks if chunk)
return extracted_text
except Exception as e:
print(f"HTML抽出エラー: {e}")
return ""
prompt = f"""
あなたは気象災害の専門家です。以下の報告書テキストから、災害事例データベース用の情報を抽出してください。
【厳守するルール】
1. テキスト内の日時は日本標準時（JST）です。気象データとの同期エラーを防ぐため、出力するJSON内の日時は「必ずUTC（協定世界時）に変換して」ISO8601形式で出力してください。
2. 和暦から西暦への変換に注意してください。現在は2026年（令和8年）です。（例: 令和4年=2022年、令和5年=2023年、令和6年=2024年、令和7年=2025年、令和8年=2026年）。2030年代など未来の日付を出力しないでください。
3. 警戒レベル（alert_level）は、死者や大規模な立ち往生、都市機能停止がある場合は「5」、運休・通行止め等は「4」、遅延は「3」、その他は「2」として判定してください。
（中略：スキーマ定義はそのまま）
# メイン処理のループ内での呼び出しイメージ
if pdf_url.lower().endswith('.pdf'):
raw_text = extract_text_from_pdf_url(pdf_url)
else:
raw_text = extract_text_from_html_url(pdf_url)
