135 lines
3.4 KiB
Python
135 lines
3.4 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
AZA Dev Status – lightweight console viewer.
|
||
|
||
Polls /api/project/status every 5 seconds and displays
|
||
the current project state plus the last 15 history entries.
|
||
Exit cleanly with Ctrl+C.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import sys
|
||
import time
|
||
from pathlib import Path
|
||
from urllib.request import Request, urlopen
|
||
from urllib.error import URLError
|
||
|
||
_BASE_DIR = Path(__file__).resolve().parent.parent
|
||
_TOKEN_FILE = _BASE_DIR / "backend_token.txt"
|
||
_HISTORY_FILE = _BASE_DIR / "project_status_history.jsonl"
|
||
_API_URL = "http://127.0.0.1:8000/api/project/status"
|
||
_REFRESH = 5
|
||
|
||
|
||
def _read_token() -> str:
|
||
tok = os.environ.get("MEDWORK_API_TOKEN", "").strip()
|
||
if tok:
|
||
return tok
|
||
try:
|
||
with open(str(_TOKEN_FILE), "r", encoding="utf-8") as f:
|
||
return (f.readline() or "").strip()
|
||
except Exception:
|
||
return ""
|
||
|
||
|
||
def _fetch_status(token: str) -> dict | None:
|
||
try:
|
||
req = Request(_API_URL, headers={"X-API-Token": token})
|
||
with urlopen(req, timeout=4) as resp:
|
||
return json.loads(resp.read().decode("utf-8", errors="replace"))
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def _read_history(n: int = 15) -> list[str]:
|
||
try:
|
||
with open(str(_HISTORY_FILE), "r", encoding="utf-8") as f:
|
||
lines = f.readlines()
|
||
return lines[-n:]
|
||
except Exception:
|
||
return []
|
||
|
||
|
||
def _clear():
|
||
os.system("cls" if os.name == "nt" else "clear")
|
||
|
||
|
||
def _render(data: dict | None, history: list[str]):
|
||
_clear()
|
||
print("=" * 60)
|
||
print(" AZA Dev Status")
|
||
print("=" * 60)
|
||
|
||
if data is None:
|
||
print("\n Waiting for backend ...\n")
|
||
else:
|
||
fields = [
|
||
("Project", data.get("project", "")),
|
||
("Phase", data.get("phase", "")),
|
||
("Current Step", data.get("current_step", "")),
|
||
("Next Step", data.get("next_step", "")),
|
||
("Last Update", data.get("last_update", "")),
|
||
("Updated At", data.get("updated_at", "")),
|
||
]
|
||
print()
|
||
for label, val in fields:
|
||
print(f" {label:14s}: {val}")
|
||
|
||
notes = data.get("notes", "")
|
||
if notes:
|
||
print(f"\n Notes: {notes}")
|
||
|
||
print()
|
||
print("-" * 60)
|
||
print(" History (last 15)")
|
||
print("-" * 60)
|
||
|
||
if not history:
|
||
print(" (no history yet)")
|
||
else:
|
||
for line in history:
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
try:
|
||
entry = json.loads(line)
|
||
ts = entry.get("ts_utc", "")[:19].replace("T", " ")
|
||
phase = entry.get("phase", "")
|
||
step = entry.get("current_step", "")
|
||
print(f" {ts} {phase:12s} {step}")
|
||
except Exception:
|
||
pass
|
||
|
||
print()
|
||
print(f" Refresh every {_REFRESH}s | Ctrl+C to exit")
|
||
print("=" * 60)
|
||
|
||
|
||
def main():
|
||
if os.name == "nt":
|
||
try:
|
||
os.system("title AZA Dev Status")
|
||
except Exception:
|
||
pass
|
||
|
||
token = _read_token()
|
||
if not token:
|
||
print("WARNING: No API token found. Requests will fail (401).")
|
||
|
||
try:
|
||
while True:
|
||
data = _fetch_status(token)
|
||
history = _read_history()
|
||
_render(data, history)
|
||
time.sleep(_REFRESH)
|
||
except KeyboardInterrupt:
|
||
print("\n Bye.")
|
||
sys.exit(0)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|