nuiks-sysmon/main.py

from fastapi import FastAPI
import psutil
import time
import datetime
from pydantic import BaseModel
from sqlalchemy import create_engine, Column, Integer, String, DateTime
from sqlalchemy.orm import sessionmaker, declarative_base
from fastapi import HTTPException
import os
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
import threading

if os.environ.get("DATABASE_URL") == None:
    os.environ["DATABASE_URL"] = "sqlite:///./wall_messages.db"
else:
    DATABASE_URL = os.environ["DATABASE_URL"]
    
    
    
app = FastAPI()

cpu_normal = True
ram_normal = True
disk_normal = True



engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine)
Base = declarative_base()

class LimitsModel(Base):
    __tablename__ = "limits"
    id = Column(Integer, primary_key=True, index=True)
    cpu_threshold = Column(Integer, default=80)
    ram_threshold = Column(Integer, default=80)
    disk_threshold = Column(Integer, default=80)

class OverageEventModel(Base):
    __tablename__ = "overages"
    id = Column(Integer, primary_key=True, index=True)
    cpu_overage = Column(Integer, default=0)
    ram_overage = Column(Integer, default=0)
    disk_overage = Column(Integer, default=0)
    timestamp = Column(DateTime, default=datetime.datetime.now(), nullable=False)

class OverageEvent(BaseModel):
    id: int
    cpu_overage: int
    ram_overage: int
    disk_overage: int
    timestamp: datetime.datetime

    class Config:
        orm_mode = True

Base.metadata.create_all(bind=engine)


@app.on_event("startup")
def startup_event():
    db = SessionLocal()
    if not db.query(LimitsModel).first():
        limits = LimitsModel(
            cpu_threshold=50,
            ram_threshold=50,
            disk_threshold=80
        )
        db.add(limits)
        db.commit()
    db.close()

def check_and_log_overages():
    while True:
        db = SessionLocal()
        limits = db.query(LimitsModel).first()
        if limits:
            cpu_limit = limits.cpu_threshold
            ram_limit = limits.ram_threshold
            disk_limit = limits.disk_threshold
        print("LIMITS: ",cpu_limit, ram_limit, disk_limit)
        cpu = psutil.cpu_percent(interval=0.5)
        ram = psutil.virtual_memory().percent
        disk = psutil.disk_usage('/').percent

        cpu_over = int(cpu > cpu_limit)
        ram_over = int(ram > ram_limit)
        disk_over = int(disk > disk_limit)
        
        last_event = db.query(OverageEventModel).order_by(OverageEventModel.timestamp.desc()).first()
        should_log = False

        if last_event is None:
            should_log = True
        else:
            if (
                last_event.cpu_overage != cpu_over or
                last_event.ram_overage != ram_over or
                last_event.disk_overage != disk_over
            ):
                should_log = True

        if should_log:
            event = OverageEventModel(
                cpu_overage=cpu_over,
                ram_overage=ram_over,
                disk_overage=disk_over,
                timestamp=datetime.datetime.now()
            )
            db.add(event)
            db.commit()
        db.close()
        time.sleep(10)

threading.Thread(target=check_and_log_overages, daemon=True).start()

start_time = time.time()

class Info(BaseModel):
    uptime: datetime.datetime
    cpu: float
    ram: float
    disk: float
    cpu_threshold: int
    ram_threshold: int
    disk_threshold: int
    cpu_normal: bool
    ram_normal: bool
    disk_normal: bool
    
class Thresholds(BaseModel):
    cpu_threshold: int
    ram_threshold: int
    disk_threshold: int


@app.get("/info", response_model=Info)
def get_info():
    db = SessionLocal()
    limits = db.query(LimitsModel).first()
    if not limits:
        db.close()
        raise HTTPException(status_code=404, detail="Limits not found")
    delta_seconds = int(time.time() - start_time)
    system_information = Info(
        uptime=datetime.datetime.now() - datetime.timedelta(seconds=delta_seconds),
        cpu=psutil.cpu_percent(interval=0.5),
        ram=psutil.virtual_memory().percent,
        disk=psutil.disk_usage('/').percent,
        cpu_threshold=limits.cpu_threshold,
        ram_threshold=limits.ram_threshold,
        disk_threshold=limits.disk_threshold,
        cpu_normal=cpu_normal,
        ram_normal=ram_normal,
        disk_normal=disk_normal
    )
    print(system_information)
    return system_information

@app.get("/limits", response_model=Thresholds)
def get_limits():
    db = SessionLocal()
    limits = db.query(LimitsModel).first()
    if not limits:
        db.close()
        raise HTTPException(status_code=404, detail="Limits not found")
    return {
        "cpu_threshold": limits.cpu_threshold,
        "ram_threshold": limits.ram_threshold,
        "disk_threshold": limits.disk_threshold
    }

@app.post("/limits")
def set_limits(thresholds: Thresholds):
    print (thresholds)
    db = SessionLocal()
    limits = db.query(LimitsModel).first()
    cpu_threshold = thresholds.cpu_threshold
    ram_threshold = thresholds.ram_threshold
    disk_threshold = thresholds.disk_threshold
    if limits:
        limits.cpu_threshold = cpu_threshold
        limits.ram_threshold = ram_threshold
        limits.disk_threshold = disk_threshold
        db.commit()
    db.close()
    return {
        "cpu": cpu_threshold,
        "ram": ram_threshold,
        "disk": disk_threshold
    }
@app.get("/overages", response_model=list[OverageEvent])
def get_overages():
    db = SessionLocal()
    overages = db.query(OverageEventModel).order_by(OverageEventModel.timestamp.desc()).all()
    db.close()
    return overages

@app.get("/overages/{overage_id}", response_model=OverageEvent)
def get_overage(overage_id: int):
    db = SessionLocal()
    overage = db.query(OverageEventModel).filter(OverageEventModel.id == overage_id).first()
    db.close()
    if not overage:
        raise HTTPException(status_code=404, detail="Overage event not found")
    return overage

@app.delete("/overages/{overage_id}")
def delete_overage(overage_id: int):
    db = SessionLocal()
    overage = db.query(OverageEventModel).filter(OverageEventModel.id == overage_id).first()
    if not overage:
        db.close()
        raise HTTPException(status_code=404, detail="Overage event not found")
    db.delete(overage)
    db.commit()
    db.close()
    return {"detail": "Overage event deleted"}

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

app.mount("/", StaticFiles(directory="./webroot", html=True), name="static")