import gradio as gr import requests import pandas as pd from dotenv import load_dotenv import os load_dotenv() # --- 環境変数と定数の設定 --- # NewsAPI設定 NEWS_API_URL = os.getenv("NEWS_API_URL", "https://newsapi.org/v2/everything") NEWS_API_KEY = os.getenv("API_KEY") # Alpha Vantage API設定 ALPHA_VANTAGE_API_URL = "https://www.alphavantage.co/query" ALPHA_VANTAGE_API_KEY = os.getenv("ALPHA_VANTAGE_API_KEY") # 環境変数から読み込む # --- ニュース検索機能 --- def fetch_news(query="ecolab", sort_by="publishedAt", page_size=10, language="auto"): """ NewsAPIからニュース情報を取得する関数 """ if not NEWS_API_KEY: return pd.DataFrame(), "❌ NewsAPIのAPIキーが設定されていません。" try: headers = { "X-API-Key": NEWS_API_KEY, "User-Agent": "NewsApp/1.0" } params = { "q": query, "sortBy": sort_by, "pageSize": page_size, } # 言語設定 if language in ["en", "jp"]: params["language"] = language elif language == "auto" and not any('\u3040' <= char <= '\u309F' or '\u30A0' <= char <= '\u30FF' or '\u4E00' <= char <= '\u9FAF' for char in query): params["language"] = "en" response = requests.get(NEWS_API_URL, headers=headers, params=params) response.raise_for_status() # エラーがあれば例外を発生させる data = response.json() if data["status"] == "ok" and data["totalResults"] > 0: articles = data["articles"] news_data = [{ "タイトル": article.get("title"), "説明": article.get("description"), "ソース": article.get("source", {}).get("name"), "公開日": article.get("publishedAt"), "URL": article.get("url") } for article in articles] df = pd.DataFrame(news_data) return df, f"✅ {len(articles)}件のニュースを取得しました" else: return pd.DataFrame(), "❌ ニュースが見つかりませんでした" except requests.exceptions.RequestException as e: return pd.DataFrame(), f"❌ APIリクエストエラー: {e}" except Exception as e: return pd.DataFrame(), f"❌ 不明なエラーが発生しました: {e}" def search_news(query: str, sort_by: str, page_size: int, language: str): if not query.strip(): return pd.DataFrame(), "❌ 検索キーワードを入力してください" return fetch_news(query, sort_by, page_size, language) # --- 暗号資産情報取得機能 --- def fetch_crypto_data(function, symbol, market): """ Alpha Vantage APIから暗号資産の情報を取得する関数 """ if not ALPHA_VANTAGE_API_KEY: return pd.DataFrame(), "❌ Alpha VantageのAPIキーが設定されていません。" if not symbol.strip() or not market.strip(): return pd.DataFrame(), "❌ 通貨シンボルと市場を両方入力してください。" params = { "apikey": ALPHA_VANTAGE_API_KEY, "function": function } # functionに応じてパラメータを動的に設定 if function == "CURRENCY_EXCHANGE_RATE": params["from_currency"] = symbol.strip() params["to_currency"] = market.strip() else: params["symbol"] = symbol.strip() params["market"] = market.strip() try: response = requests.get(ALPHA_VANTAGE_API_URL, params=params) response.raise_for_status() data = response.json() # APIからのエラーメッセージをチェック if "Error Message" in data: return pd.DataFrame(), f"❌ APIエラー: {data['Error Message']}" if "Information" in data: return pd.DataFrame(), f"ℹ️ API情報: {data['Information']}" if not data: return pd.DataFrame(), "❌ APIから有効なデータが返されませんでした。" # 取得したデータに応じてDataFrameに変換 if function == "CURRENCY_EXCHANGE_RATE": key = "Realtime Currency Exchange Rate" if key in data: # データを整形してDataFrameに rate_data = { "項目": list(data[key].keys()), "値": list(data[key].values()) } df = pd.DataFrame(rate_data) return df, f"✅ {symbol}/{market}のリアルタイム価格を取得しました。" else: return pd.DataFrame(), "❌ リアルタイム価格のデータを取得できませんでした。" else: # 時系列データ (DAILY, WEEKLY, MONTHLY) # "Time Series (...)" というキーを探す ts_key = next((k for k in data.keys() if k.startswith("Time Series")), None) if ts_key and data[ts_key]: df = pd.DataFrame.from_dict(data[ts_key], orient='index') df.index.name = "日付" # カラム名を分かりやすく変更 df.columns = [col.split('. ')[1] for col in df.columns] return df, f"✅ {symbol}/{market}の{function.replace('DIGITAL_CURRENCY_', '')}データを取得しました。" else: return pd.DataFrame(), "❌ 時系列データを取得できませんでした。シンボルや市場が正しいか確認してください。" except requests.exceptions.RequestException as e: return pd.DataFrame(), f"❌ APIリクエストエラー: {e}" except Exception as e: return pd.DataFrame(), f"❌ 不明なエラーが発生しました: {e}" # --- Gradioインターフェース --- with gr.Blocks(title="📰 総合情報検索アプリ", theme=gr.themes.Soft()) as demo: gr.Markdown("# 📰 総合情報検索アプリ") gr.Markdown("ニュース検索と暗号資産の価格情報を取得できます。") with gr.Tabs(): # --- ニュース検索タブ --- with gr.TabItem("📰 ニュース検索"): gr.Markdown( """ **使い方:** 1. 検索キーワードを入力（日本語・英語両方対応） 2. 言語とソート方法を選択 3. 取得件数を選択 4. 「ニュース検索」ボタンをクリック """ ) with gr.Row(): with gr.Column(scale=1): news_query_input = gr.Textbox( label="🔍 検索キーワード", placeholder="例: 東京, トヨタ, AI, technology...", value="Ecolab" ) language_dropdown = gr.Dropdown( choices=[("自動検出", "auto"), ("日本語", "jp"), ("英語", "en"), ("すべての言語", "all")], label="🌐 言語", value="auto" ) sort_by_dropdown = gr.Dropdown( choices=["publishedAt", "relevancy", "popularity"], label="📊 ソート方法", value="publishedAt" ) page_size_slider = gr.Slider( minimum=5, maximum=50, value=10, step=5, label="📄 取得件数" ) news_search_btn = gr.Button("🔍 ニュース検索", variant="primary") with gr.Column(scale=3): news_status_output = gr.Textbox(label="📋 ステータス", interactive=False) news_results_df = gr.DataFrame(label="📰 ニュース結果", wrap=True, interactive=False) # --- 暗号資産検索タブ --- with gr.TabItem("🪙 暗号資産検索"): gr.Markdown( """ **使い方:** 1. 取得したいデータの種類を選択 2. 暗号資産のシンボルを入力 (例: `BTC`, `ETH`) 3. 取引市場の通貨を入力 (例: `USD`, `EUR`) 4. 「データ取得」ボタンをクリック **注:** `JPY`などの一部市場は、時系列データ（日足・週足・月足）でエラーになる場合があります。その際は`USD`など主要な市場をお試しください。 """ ) with gr.Row(): with gr.Column(scale=1): crypto_function_dropdown = gr.Dropdown( label="📊 データ種別", choices=[ ("日足データ", "DIGITAL_CURRENCY_DAILY"), ("週足データ", "DIGITAL_CURRENCY_WEEKLY"), ("月足データ", "DIGITAL_CURRENCY_MONTHLY"), ("リアルタイム価格", "CURRENCY_EXCHANGE_RATE") ], value="DIGITAL_CURRENCY_DAILY" ) crypto_symbol_input = gr.Textbox( label="🪙 通貨シンボル", placeholder="例: BTC", value="BTC" ) crypto_market_input = gr.Textbox( label="💴 取引市場", placeholder="例: USD", value="USD" ) crypto_search_btn = gr.Button("🔍 データ取得", variant="primary") with gr.Column(scale=3): crypto_status_output = gr.Textbox(label="📋 ステータス", interactive=False) crypto_results_df = gr.DataFrame(label="📈 暗号資産データ", wrap=True, interactive=False) # --- イベントハンドラー --- # ニュース検索 news_search_btn.click( fn=search_news, inputs=[news_query_input, sort_by_dropdown, page_size_slider, language_dropdown], outputs=[news_results_df, news_status_output] ) news_query_input.submit( fn=search_news, inputs=[news_query_input, sort_by_dropdown, page_size_slider, language_dropdown], outputs=[news_results_df, news_status_output] ) # 暗号資産検索 crypto_search_btn.click( fn=fetch_crypto_data, inputs=[crypto_function_dropdown, crypto_symbol_input, crypto_market_input], outputs=[crypto_results_df, crypto_status_output] ) # Enterキーでの実行（Textboxのみ） for inp in [crypto_symbol_input, crypto_market_input]: inp.submit( fn=fetch_crypto_data, inputs=[crypto_function_dropdown, crypto_symbol_input, crypto_market_input], outputs=[crypto_results_df, crypto_status_output] ) # 初期ロード時のデータ表示 demo.load( fn=lambda: search_news("Ecolab", "publishedAt", 10, "auto"), outputs=[news_results_df, news_status_output] ) # アプリ起動 if __name__ == "__main__": demo.launch()