Started repo and finished the bot in one shot

2026-08-16 16:37:01 -05:00
commit 9788c5d9b3
8 changed files with 303 additions and 0 deletions
@@ -0,0 +1,90 @@
+import asyncio
+import json
+import time
+
+import aiohttp
+
+TARGETS = [
+    {
+        'name': 'Gitea',
+        'url': 'https://gitea.rainbow-herring.ts.net/api/healthz',
+        'kind': 'gitea',
+    },
+    {
+        'name': 'Packs Site (Production)',
+        'url': 'https://jakerasp.rainbow-herring.ts.net/healthz/',
+        'kind': 'packs',
+    },
+    {
+        'name': 'Packs Site (Development)',
+        'url': 'https://jakepc.rainbow-herring.ts.net/healthz/',
+        'kind': 'packs',
+    },
+]
+
+GITEA_VERSION_URL = 'https://gitea.rainbow-herring.ts.net/api/v1/version'
+
+
+async def _get(session, url, timeout):
+    started = time.monotonic()
+    async with session.get(
+        url,
+        timeout=aiohttp.ClientTimeout(total=timeout),
+        allow_redirects=False,
+    ) as resp:
+        latency_ms = round((time.monotonic() - started) * 1000)
+        body = await resp.text()
+        return resp.status, latency_ms, body
+
+
+async def check_target(session, target, timeout):
+    result = {
+        'name': target['name'],
+        'kind': target.get('kind', ''),
+        'url': target['url'],
+        'up': False,
+        'status_code': None,
+        'latency_ms': None,
+        'db': None,
+        'version': None,
+        'detail': '',
+    }
+    try:
+        status, latency_ms, body = await _get(session, target['url'], timeout)
+        result['status_code'] = status
+        result['latency_ms'] = latency_ms
+
+        if status == 200:
+            data = {}
+            try:
+                data = json.loads(body)
+            except (json.JSONDecodeError, ValueError):
+                pass
+            result['up'] = data.get('status') == 'pass'
+            if data.get('db') is not None:
+                result['db'] = data['db']
+        else:
+            result['detail'] = f'HTTP {status}'
+    except asyncio.TimeoutError:
+        result['detail'] = 'timeout'
+    except aiohttp.ClientError:
+        result['detail'] = 'connection error'
+    except Exception as exc:
+        result['detail'] = type(exc).__name__
+
+    if result['up'] and result['kind'] == 'gitea':
+        try:
+            status, _, body = await _get(session, GITEA_VERSION_URL, timeout)
+            if status == 200:
+                result['version'] = json.loads(body).get('version')
+        except Exception:
+            pass
+
+    return result
+
+
+async def run_checks(session, timeout):
+    results = await asyncio.gather(
+        *(check_target(session, target, timeout) for target in TARGETS)
+    )
+    return list(results)