scripts/nostur_status_query.py

@@ -0,0 +1,89 @@
+#!/usr/bin/env python3
+"""
+Nostur Status Query MVP
+Read-only status responses sourced from Gitea truth.
+"""
+import json
+import os
+import sys
+import urllib.request
+from datetime import datetime
+
+# Configuration
+GITEA_URL = os.environ.get("GITEA_URL", "https://forge.alexanderwhitestone.com/api/v1")
+GITEA_TOKEN = os.environ.get("GITEA_TOKEN", "f7bcdaf878d479ad7747873ff6739a9bb89e3f80")
+REPO_OWNER = "Timmy_Foundation"
+REPO_NAME = "timmy-config"
+
+def gitea_get(path):
+    if not GITEA_TOKEN:
+        raise RuntimeError("GITEA_TOKEN not set")
+    
+    url = f"{GITEA_URL}{path}"
+    headers = {"Authorization": f"token {GITEA_TOKEN}"}
+    req = urllib.request.Request(url, headers=headers)
+    
+    try:
+        with urllib.request.urlopen(req, timeout=15) as resp:
+            return json.loads(resp.read().decode())
+    except Exception as e:
+        print(f"Error fetching from Gitea: {e}", file=sys.stderr)
+        return None
+
+def get_status():
+    path = f"/repos/{REPO_OWNER}/{REPO_NAME}/issues?state=open&limit=50"
+    issues = gitea_get(path)
+    
+    if not issues:
+        return "⚠️ Error: Could not fetch status from Gitea."
+
+    # 1. Active Epics
+    active_epics = [
+        i['title'].replace('[EPIC]', '').strip() 
+        for i in issues 
+        if '[EPIC]' in i['title'] or any(l['name'] == 'epic' for l in i.get('labels', []))
+    ][:2]
+
+    # 2. Blockers / Critical (Bugs, Security, Ops)
+    blockers = [
+        i['title'].strip()
+        for i in issues
+        if any(tag in i['title'] for tag in ['[BUG]', '[SECURITY]', '[OPS]']) or any(l['name'] == 'blocker' for l in i.get('labels', []))
+    ][:2]
+
+    # 3. Priority Queue (Top 3)
+    priority_queue = [
+        f"#{i['number']}: {i['title']}"
+        for i in issues[:3]
+    ]
+
+    # Format compact response for mobile
+    now = datetime.now().strftime("%H:%M:%S")
+    
+    lines = ["[TIMMY STATUS]"]
+    lines.append(f"EPICS: {' | '.join(active_epics) if active_epics else 'None'}")
+    lines.append(f"BLOCKERS: {' | '.join(blockers) if blockers else 'None'}")
+    lines.append(f"QUEUE: {' | '.join(priority_queue)}")
+    lines.append(f"UPDATED: {now}")
+    
+    return "\n".join(lines)
+
+if __name__ == "__main__":
+    # If called with 'json', return JSON payload
+    if len(sys.argv) > 1 and sys.argv[1] == "--json":
+        path = f"/repos/{REPO_OWNER}/{REPO_NAME}/issues?state=open&limit=50"
+        issues = gitea_get(path)
+        if not issues:
+            print(json.dumps({"error": "fetch failed"}))
+            sys.exit(1)
+            
+        data = {
+            "status": "active",
+            "epics": [i['title'] for i in issues if '[EPIC]' in i['title']][:2],
+            "blockers": [i['title'] for i in issues if any(tag in i['title'] for tag in ['[BUG]', '[SECURITY]', '[OPS]'])][:2],
+            "queue": [f"#{i['number']}: {i['title']}" for i in issues[:3]],
+            "timestamp": datetime.now().isoformat()
+        }
+        print(json.dumps(data, indent=2))
+    else:
+        print(get_status())