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()
|