| 109 | |
| 110 | @app.get("/live") |
| 111 | async def live_status(): |
| 112 | import time |
| 113 | now = time.monotonic() |
| 114 | if now - _live_cache["last_checked"] < LIVE_CACHE_TTL: |
| 115 | return {"live": _live_cache["live"], "videoId": _live_cache["videoId"]} |
| 116 | |
| 117 | api_key = os.environ.get("YOUTUBE_API_KEY") |
| 118 | if not api_key: |
| 119 | return {"live": False, "videoId": None} |
| 120 | |
| 121 | url = "https://www.googleapis.com/youtube/v3/search" |
| 122 | params = { |
| 123 | "part": "id", |
| 124 | "channelId": YOUTUBE_CHANNEL_ID, |
| 125 | "eventType": "live", |
| 126 | "type": "video", |
| 127 | "key": api_key, |
| 128 | } |
| 129 | async with httpx.AsyncClient() as client: |
| 130 | resp = await client.get(url, params=params, timeout=10) |
| 131 | |
| 132 | if resp.status_code != 200: |
| 133 | _live_cache["last_checked"] = now |
| 134 | return {"live": False, "videoId": None} |
| 135 | |
| 136 | items = resp.json().get("items", []) |
| 137 | result = {"live": bool(items), "videoId": items[0]["id"]["videoId"] if items else None} |
| 138 | _live_cache.update({**result, "last_checked": now}) |
| 139 | return result |
| 140 | |
| 141 | |
| 142 | # Root path to check if service is running |