"""EIA RSS feed HTML widget.""" from typing import Annotated from fastapi import Depends from fastapi.responses import HTMLResponse from openbb_core.app.router import Router from openbb_us_eia.browsers import _mcp_url, request_info from openbb_us_eia.utils.rss import ( EIA_RSS_FEEDS, build_feed, feed_choices, render_rss_html, ) router = Router(prefix="", description="EIA RSS feeds.") async def rss_feed( info: Annotated[dict, Depends(request_info)], feed: str = "today_in_energy", theme: str = "dark", ) -> HTMLResponse: """Articles, releases, testimony, and fuel-price updates from EIA's RSS feeds.""" from openbb_core.provider.utils.helpers import get_async_requests_session from openbb_us_eia.utils.rss import fetch_feed if feed not in EIA_RSS_FEEDS: feed = "today_in_energy" proxy_base = info["url"].rsplit("/rss_feed", 1)[0] + "/eia_proxy" async with await get_async_requests_session() as session: parsed = await fetch_feed(session, EIA_RSS_FEEDS[feed]["url"]) data = build_feed(feed, parsed, 50, proxy_base) data["endpoint"] = info["url"] return HTMLResponse(content=render_rss_html(data, theme)) router._api_router.add_api_route( path="/rss_feed", endpoint=rss_feed, methods=["GET"], response_class=HTMLResponse, openapi_extra={ "widget_config": { "name": "EIA RSS Feeds", "description": "Articles, releases, testimony, and fuel-price updates" " from the U.S. Energy Information Administration RSS feeds.", "category": "EIA", "subCategory": "News & Feeds", "source": ["EIA"], "type": "iframe", "widgetId": "eia_rss_feeds_us_eia_obb", "gridData": {"w": 40, "h": 20}, "params": [ { "paramName": "feed", "label": "Feed", "type": "tabs", "value": "today_in_energy", "description": "The EIA RSS feed to display.", "show": True, "options": feed_choices(), }, {"paramName": "theme", "show": False}, ], "refetchInterval": 900000, "storage": {"mcpUrl": _mcp_url()}, } }, )