310 lines
11 KiB
Python
310 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""JAESWIFT HUD Backend API"""
|
|
import json, os, time, subprocess, random, datetime, hashlib
|
|
from functools import wraps
|
|
from pathlib import Path
|
|
|
|
from flask import Flask, request, jsonify, abort
|
|
from flask_cors import CORS
|
|
import jwt
|
|
import requests as req
|
|
|
|
app = Flask(__name__)
|
|
CORS(app)
|
|
|
|
DATA_DIR = Path(__file__).parent / 'data'
|
|
JWT_SECRET = 'jaeswift-hud-s3cr3t-2026!x'
|
|
ADMIN_USER = 'jae'
|
|
ADMIN_PASS = 'HUDAdmin2026!'
|
|
|
|
# ─── Helpers ─────────────────────────────────────────
|
|
def load_json(name):
|
|
p = DATA_DIR / name
|
|
if p.exists():
|
|
with open(p) as f:
|
|
return json.load(f)
|
|
return [] if name.endswith('posts.json') else {}
|
|
|
|
def save_json(name, data):
|
|
p = DATA_DIR / name
|
|
p.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(p, 'w') as f:
|
|
json.dump(data, f, indent=2)
|
|
|
|
def require_auth(fn):
|
|
@wraps(fn)
|
|
def wrapper(*a, **kw):
|
|
auth = request.headers.get('Authorization', '')
|
|
if not auth.startswith('Bearer '):
|
|
abort(401, 'Missing token')
|
|
try:
|
|
jwt.decode(auth[7:], JWT_SECRET, algorithms=['HS256'])
|
|
except Exception:
|
|
abort(401, 'Invalid token')
|
|
return fn(*a, **kw)
|
|
return wrapper
|
|
|
|
def shell(cmd):
|
|
try:
|
|
r = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=5)
|
|
return r.stdout.strip()
|
|
except Exception:
|
|
return ''
|
|
|
|
# ─── Auth ────────────────────────────────────────────
|
|
@app.route('/api/auth/login', methods=['POST'])
|
|
def login():
|
|
d = request.get_json(force=True, silent=True) or {}
|
|
if d.get('username') == ADMIN_USER and d.get('password') == ADMIN_PASS:
|
|
token = jwt.encode(
|
|
{'user': ADMIN_USER, 'exp': datetime.datetime.utcnow() + datetime.timedelta(hours=24)},
|
|
JWT_SECRET, algorithm='HS256'
|
|
)
|
|
return jsonify({'token': token})
|
|
return jsonify({'error': 'Invalid credentials'}), 401
|
|
|
|
@app.route('/api/auth/check')
|
|
def auth_check():
|
|
auth = request.headers.get('Authorization', '')
|
|
if not auth.startswith('Bearer '):
|
|
return jsonify({'valid': False}), 401
|
|
try:
|
|
jwt.decode(auth[7:], JWT_SECRET, algorithms=['HS256'])
|
|
return jsonify({'valid': True, 'user': ADMIN_USER})
|
|
except Exception:
|
|
return jsonify({'valid': False}), 401
|
|
|
|
# ─── Blog Posts ──────────────────────────────────────
|
|
@app.route('/api/posts')
|
|
def get_posts():
|
|
posts = load_json('posts.json')
|
|
return jsonify(posts)
|
|
|
|
@app.route('/api/posts/<slug>')
|
|
def get_post(slug):
|
|
posts = load_json('posts.json')
|
|
for p in posts:
|
|
if p.get('slug') == slug:
|
|
return jsonify(p)
|
|
abort(404)
|
|
|
|
@app.route('/api/posts', methods=['POST'])
|
|
@require_auth
|
|
def create_post():
|
|
d = request.get_json(force=True)
|
|
posts = load_json('posts.json')
|
|
d['id'] = max((p.get('id', 0) for p in posts), default=0) + 1
|
|
d['date'] = d.get('date', datetime.date.today().isoformat())
|
|
d['word_count'] = len(d.get('content', '').split())
|
|
posts.append(d)
|
|
save_json('posts.json', posts)
|
|
return jsonify(d), 201
|
|
|
|
@app.route('/api/posts/<slug>', methods=['PUT'])
|
|
@require_auth
|
|
def update_post(slug):
|
|
d = request.get_json(force=True)
|
|
posts = load_json('posts.json')
|
|
for i, p in enumerate(posts):
|
|
if p.get('slug') == slug:
|
|
d['id'] = p['id']
|
|
d['word_count'] = len(d.get('content', '').split())
|
|
posts[i] = {**p, **d}
|
|
save_json('posts.json', posts)
|
|
return jsonify(posts[i])
|
|
abort(404)
|
|
|
|
@app.route('/api/posts/<slug>', methods=['DELETE'])
|
|
@require_auth
|
|
def delete_post(slug):
|
|
posts = load_json('posts.json')
|
|
posts = [p for p in posts if p.get('slug') != slug]
|
|
save_json('posts.json', posts)
|
|
return jsonify({'ok': True})
|
|
|
|
# ─── Server Stats ────────────────────────────────────
|
|
@app.route('/api/stats')
|
|
def server_stats():
|
|
# CPU
|
|
load = shell("cat /proc/loadavg | awk '{print $1}'")
|
|
ncpu = shell("nproc")
|
|
try:
|
|
cpu_pct = round(float(load) / max(int(ncpu), 1) * 100, 1)
|
|
except Exception:
|
|
cpu_pct = 0
|
|
|
|
# Memory
|
|
mem = shell("free | awk '/Mem:/{printf \"%.1f\", $3/$2*100}'")
|
|
|
|
# Disk
|
|
disk = shell("df / | awk 'NR==2{print $5}' | tr -d '%'")
|
|
|
|
# Network (bytes since boot)
|
|
net = shell("cat /proc/net/dev | awk '/eth0|ens/{print $2,$10}'")
|
|
parts = net.split()
|
|
rx = int(parts[0]) if len(parts) >= 2 else 0
|
|
tx = int(parts[1]) if len(parts) >= 2 else 0
|
|
|
|
# Docker
|
|
running = shell("docker ps -q 2>/dev/null | wc -l")
|
|
total = shell("docker ps -aq 2>/dev/null | wc -l")
|
|
|
|
# Uptime
|
|
up = shell("cat /proc/uptime | awk '{print $1}'")
|
|
|
|
# Connections
|
|
conns = shell("ss -s | awk '/TCP:/{print $2}'")
|
|
|
|
return jsonify({
|
|
'cpu_percent': cpu_pct,
|
|
'memory_percent': float(mem) if mem else 0,
|
|
'disk_percent': int(disk) if disk else 0,
|
|
'network_rx_bytes': rx,
|
|
'network_tx_bytes': tx,
|
|
'container_running': int(running) if running else 0,
|
|
'container_total': int(total) if total else 0,
|
|
'uptime_seconds': float(up) if up else 0,
|
|
'active_connections': int(conns) if conns else 0,
|
|
'load_avg': float(load) if load else 0,
|
|
'timestamp': time.time()
|
|
})
|
|
|
|
# ─── Services Status ─────────────────────────────────
|
|
@app.route('/api/services')
|
|
def services():
|
|
svcs = [
|
|
{'name': 'Gitea', 'url': 'https://git.jaeswift.xyz'},
|
|
{'name': 'Plex', 'url': 'https://plex.jaeswift.xyz'},
|
|
{'name': 'Search', 'url': 'https://jaeswift.xyz/search'},
|
|
{'name': 'Yoink', 'url': 'https://jaeswift.xyz/yoink/'},
|
|
{'name': 'Archive', 'url': 'https://archive.jaeswift.xyz'},
|
|
{'name': 'Agent Zero', 'url': 'https://agentzero.jaeswift.xyz'},
|
|
{'name': 'Files', 'url': 'https://files.jaeswift.xyz'},
|
|
]
|
|
results = []
|
|
for s in svcs:
|
|
try:
|
|
t0 = time.time()
|
|
r = req.get(s['url'], timeout=5, verify=False, allow_redirects=True)
|
|
ms = round((time.time() - t0) * 1000)
|
|
results.append({**s, 'status': 'online' if r.status_code < 500 else 'offline', 'response_time_ms': ms})
|
|
except Exception:
|
|
results.append({**s, 'status': 'offline', 'response_time_ms': 0})
|
|
return jsonify(results)
|
|
|
|
# ─── Weather ─────────────────────────────────────────
|
|
@app.route('/api/weather')
|
|
def weather():
|
|
try:
|
|
r = req.get('https://wttr.in/Manchester?format=j1', timeout=5,
|
|
headers={'User-Agent': 'jaeswift-hud'})
|
|
d = r.json()
|
|
cur = d['current_condition'][0]
|
|
return jsonify({
|
|
'temp_c': int(cur['temp_C']),
|
|
'feels_like': int(cur['FeelsLikeC']),
|
|
'condition': cur['weatherDesc'][0]['value'],
|
|
'humidity': int(cur['humidity']),
|
|
'wind_kph': int(cur['windspeedKmph']),
|
|
'wind_dir': cur['winddir16Point'],
|
|
'icon': cur.get('weatherCode', ''),
|
|
})
|
|
except Exception as e:
|
|
return jsonify({'error': str(e)}), 500
|
|
|
|
# ─── Now Playing (random track) ──────────────────────
|
|
@app.route('/api/nowplaying')
|
|
def now_playing():
|
|
tracks = load_json('tracks.json')
|
|
if not tracks:
|
|
return jsonify({'artist': 'Unknown', 'track': 'Silence', 'album': ''})
|
|
t = random.choice(tracks)
|
|
return jsonify(t)
|
|
|
|
# ─── Tracks CRUD ─────────────────────────────────────
|
|
@app.route('/api/tracks')
|
|
def get_tracks():
|
|
return jsonify(load_json('tracks.json'))
|
|
|
|
@app.route('/api/tracks', methods=['POST'])
|
|
@require_auth
|
|
def add_track():
|
|
d = request.get_json(force=True)
|
|
tracks = load_json('tracks.json')
|
|
tracks.append({
|
|
'artist': d.get('artist', ''),
|
|
'track': d.get('track', ''),
|
|
'album': d.get('album', ''),
|
|
})
|
|
save_json('tracks.json', tracks)
|
|
return jsonify(tracks[-1]), 201
|
|
|
|
@app.route('/api/tracks/<int:index>', methods=['DELETE'])
|
|
@require_auth
|
|
def delete_track(index):
|
|
tracks = load_json('tracks.json')
|
|
if 0 <= index < len(tracks):
|
|
removed = tracks.pop(index)
|
|
save_json('tracks.json', tracks)
|
|
return jsonify({'ok': True, 'removed': removed})
|
|
abort(404)
|
|
|
|
|
|
# ─── Git Activity (from Gitea API) ───────────────────
|
|
@app.route('/api/git-activity')
|
|
def git_activity():
|
|
try:
|
|
r = req.get('https://git.jaeswift.xyz/api/v1/users/jae/heatmap',
|
|
timeout=5, verify=False)
|
|
heatmap = r.json() if r.status_code == 200 else []
|
|
|
|
r2 = req.get('https://git.jaeswift.xyz/api/v1/repos/search?sort=updated&limit=5&owner=jae',
|
|
timeout=5, verify=False)
|
|
repos = []
|
|
if r2.status_code == 200:
|
|
data = r2.json().get('data', r2.json()) if isinstance(r2.json(), dict) else r2.json()
|
|
for repo in (data if isinstance(data, list) else [])[:5]:
|
|
repos.append({
|
|
'name': repo.get('name', ''),
|
|
'updated': repo.get('updated_at', ''),
|
|
'stars': repo.get('stars_count', 0),
|
|
'language': repo.get('language', ''),
|
|
})
|
|
|
|
return jsonify({'heatmap': heatmap, 'repos': repos})
|
|
except Exception as e:
|
|
return jsonify({'heatmap': [], 'repos': [], 'error': str(e)})
|
|
|
|
# ─── Threat Feed (CVE feed) ──────────────────────────
|
|
@app.route('/api/threats')
|
|
def threats():
|
|
try:
|
|
r = req.get('https://cve.circl.lu/api/last/8', timeout=8)
|
|
cves = []
|
|
if r.status_code == 200:
|
|
for item in r.json()[:8]:
|
|
cves.append({
|
|
'id': item.get('id', ''),
|
|
'summary': (item.get('summary', '') or '')[:120],
|
|
'published': item.get('Published', ''),
|
|
'cvss': item.get('cvss', 0),
|
|
})
|
|
return jsonify(cves)
|
|
except Exception:
|
|
return jsonify([])
|
|
|
|
# ─── Settings ────────────────────────────────────────
|
|
@app.route('/api/settings')
|
|
def get_settings():
|
|
return jsonify(load_json('settings.json'))
|
|
|
|
@app.route('/api/settings', methods=['PUT'])
|
|
@require_auth
|
|
def update_settings():
|
|
d = request.get_json(force=True)
|
|
save_json('settings.json', d)
|
|
return jsonify(d)
|
|
|
|
# ─── Run ─────────────────────────────────────────────
|
|
if __name__ == '__main__':
|
|
app.run(host='0.0.0.0', port=5000, debug=False)
|