156 lines
4.8 KiB
Python
156 lines
4.8 KiB
Python
from fastapi import APIRouter, Request, Depends, HTTPException, Query
|
|
from fastapi.responses import HTMLResponse, RedirectResponse
|
|
from fastapi.templating import Jinja2Templates
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import desc
|
|
from datetime import datetime, timezone, timedelta
|
|
from typing import Optional
|
|
|
|
from database import get_db
|
|
from auth import verify_token
|
|
from models import Computer, Heartbeat, User, Notification
|
|
from notifications import check_offline_computers, get_unread_count
|
|
|
|
router = APIRouter()
|
|
templates = Jinja2Templates(directory="templates")
|
|
|
|
|
|
def user_or_redirect(request: Request, db: Session):
|
|
token = request.cookies.get("access_token")
|
|
if not token:
|
|
return None
|
|
payload = verify_token(token, "access")
|
|
if not payload or "sub" not in payload:
|
|
return None
|
|
return db.query(User).filter(User.username == payload["sub"]).first()
|
|
|
|
|
|
@router.get("/", response_class=HTMLResponse)
|
|
def index(
|
|
request: Request,
|
|
status: Optional[str] = Query(None),
|
|
user: Optional[str] = Query(None),
|
|
search: Optional[str] = Query(None),
|
|
sort: str = Query("hostname"),
|
|
order: str = Query("asc"),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
current_user = user_or_redirect(request, db)
|
|
if not current_user:
|
|
return RedirectResponse(url="/login")
|
|
|
|
check_offline_computers(db)
|
|
|
|
query = db.query(Computer)
|
|
if status and status in ("online", "offline"):
|
|
query = query.filter(Computer.status == status)
|
|
if user:
|
|
query = query.filter(Computer.current_user.ilike(f"%{user}%"))
|
|
if search:
|
|
query = query.filter(
|
|
Computer.hostname.ilike(f"%{search}%")
|
|
| Computer.os_info.ilike(f"%{search}%")
|
|
| Computer.current_ip.ilike(f"%{search}%")
|
|
)
|
|
|
|
allowed_sort = {"hostname", "last_seen", "current_user", "status", "current_cpu_percent", "current_ram_percent"}
|
|
if sort in allowed_sort:
|
|
column = getattr(Computer, sort)
|
|
if order == "desc":
|
|
query = query.order_by(desc(column))
|
|
else:
|
|
query = query.order_by(column)
|
|
else:
|
|
query = query.order_by(Computer.hostname)
|
|
|
|
computers = query.all()
|
|
|
|
return templates.TemplateResponse(
|
|
"dashboard.html",
|
|
{
|
|
"request": request,
|
|
"user": current_user,
|
|
"computers": computers,
|
|
"filters": {
|
|
"status": status or "",
|
|
"user": user or "",
|
|
"search": search or "",
|
|
"sort": sort,
|
|
"order": order,
|
|
},
|
|
"unread_count": get_unread_count(db),
|
|
"now": datetime.now(timezone.utc),
|
|
},
|
|
)
|
|
|
|
|
|
@router.get("/login", response_class=HTMLResponse)
|
|
def login_page(request: Request):
|
|
return templates.TemplateResponse("login.html", {"request": request, "error": None})
|
|
|
|
|
|
@router.get("/computers/{computer_id}", response_class=HTMLResponse)
|
|
def computer_page(computer_id: int, request: Request, db: Session = Depends(get_db)):
|
|
current_user = user_or_redirect(request, db)
|
|
if not current_user:
|
|
return RedirectResponse(url="/login")
|
|
|
|
computer = db.query(Computer).filter(Computer.id == computer_id).first()
|
|
if not computer:
|
|
raise HTTPException(status_code=404, detail="Computer not found")
|
|
|
|
since = datetime.now(timezone.utc) - timedelta(hours=24)
|
|
heartbeats = (
|
|
db.query(Heartbeat)
|
|
.filter(Heartbeat.computer_id == computer_id, Heartbeat.timestamp >= since)
|
|
.order_by(Heartbeat.timestamp)
|
|
.all()
|
|
)
|
|
|
|
hb_data = []
|
|
for h in heartbeats:
|
|
hb_data.append({
|
|
"timestamp": h.timestamp.isoformat() if h.timestamp else None,
|
|
"username": h.username,
|
|
"local_ip": h.local_ip,
|
|
"cpu_percent": h.cpu_percent,
|
|
"ram_percent": h.ram_percent,
|
|
"cpu_temp": h.cpu_temp,
|
|
"processes": h.processes or [],
|
|
})
|
|
|
|
return templates.TemplateResponse(
|
|
"computer.html",
|
|
{
|
|
"request": request,
|
|
"user": current_user,
|
|
"computer": computer,
|
|
"heartbeats": hb_data,
|
|
"unread_count": get_unread_count(db),
|
|
},
|
|
)
|
|
|
|
|
|
@router.get("/notifications", response_class=HTMLResponse)
|
|
def notifications_page(request: Request, db: Session = Depends(get_db)):
|
|
current_user = user_or_redirect(request, db)
|
|
if not current_user:
|
|
return RedirectResponse(url="/login")
|
|
|
|
notifications = (
|
|
db.query(Notification)
|
|
.order_by(desc(Notification.sent_at))
|
|
.limit(100)
|
|
.all()
|
|
)
|
|
|
|
return templates.TemplateResponse(
|
|
"notifications.html",
|
|
{
|
|
"request": request,
|
|
"user": current_user,
|
|
"notifications": notifications,
|
|
"unread_count": get_unread_count(db),
|
|
},
|
|
)
|