forked from Ikalus1988/MisakaNet
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathnode_status.py
More file actions
85 lines (70 loc) · 2.72 KB
/
Copy pathnode_status.py
File metadata and controls
85 lines (70 loc) · 2.72 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
#!/usr/bin/env python3
"""Node status dashboard — read MisakaNet KV to display node stats.
Usage:
# Via wrangler (requires Cloudflare auth)
python3 scripts/node_status.py
# Or pass KV namespace ID directly
python3 scripts/node_status.py --kv-id d5fb6b0797b84d17b0586fb982231ffe
Output:
Node Counter: 10060
Latest Node ID: Misaka00060
Active Nodes (sampled): 5
"""
import argparse
import json
import subprocess
import sys
from pathlib import Path
def read_counter_file(repo_root: Path) -> dict:
"""Read node counter from data/counter.json (source of truth)."""
counter_path = repo_root / "data" / "counter.json"
if not counter_path.exists():
return {"current": None, "updated": None}
try:
data = json.loads(counter_path.read_text(encoding="utf-8"))
return {"current": data.get("current"), "updated": data.get("updated", "?")[:10]}
except (json.JSONDecodeError, OSError):
return {"current": None, "updated": None}
def read_test_nodes(repo_root: Path) -> list:
"""Read test/non-formal node IDs from test-nodes.json."""
test_path = repo_root / "test-nodes.json"
if not test_path.exists():
return []
try:
data = json.loads(test_path.read_text(encoding="utf-8"))
return data.get("tests", [])
except (json.JSONDecodeError, OSError):
return []
def main():
parser = argparse.ArgumentParser(description="MisakaNet node status dashboard")
parser.add_argument("--json", action="store_true", help="JSON output")
args = parser.parse_args()
repo_root = Path(__file__).resolve().parent.parent
# Read from data/counter.json (source of truth)
counter_info = read_counter_file(repo_root)
counter = counter_info["current"]
latest_id = f"Misaka{str(counter).zfill(5)}" if counter else "unknown"
test_nodes = read_test_nodes(repo_root)
# Count lessons
lessons_dir = repo_root / "lessons"
lesson_count = 0
for subdir in ("core", "contrib"):
d = lessons_dir / subdir
if d.exists():
lesson_count += len(list(d.glob("*.md")))
if args.json:
print(json.dumps({
"counter": counter,
"latest_node_id": latest_id,
"counter_updated": counter_info["updated"],
"test_nodes_count": len(test_nodes),
"lesson_count": lesson_count,
}, ensure_ascii=False, indent=2))
else:
print(f"Node Counter: {counter or 'unknown'}")
print(f"Latest Node ID: {latest_id}")
print(f"Counter updated: {counter_info['updated']}")
print(f"Test nodes: {len(test_nodes)} (excluded from active count)")
print(f"Lessons: {lesson_count}")
if __name__ == "__main__":
main()