Update
This commit is contained in:
@@ -0,0 +1,90 @@
|
||||
"""Historique des sessions de connexion WGSecure."""
|
||||
import json
|
||||
import os
|
||||
from datetime import datetime
|
||||
from app.utils.platform_utils import get_config_dir
|
||||
|
||||
_FILE = "sessions.json"
|
||||
_MAX = 200
|
||||
|
||||
|
||||
def _path() -> str:
|
||||
return os.path.join(get_config_dir(), _FILE)
|
||||
|
||||
|
||||
def start_session(server: str, profile: str) -> int:
|
||||
"""Démarre une nouvelle session, retourne son id."""
|
||||
sid = int(datetime.now().timestamp())
|
||||
sessions = _load()
|
||||
sessions.append({
|
||||
"id": sid,
|
||||
"start": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||||
"end": None,
|
||||
"duration": None,
|
||||
"server": server,
|
||||
"profile": profile,
|
||||
"rx": 0,
|
||||
"tx": 0,
|
||||
})
|
||||
_save(sessions)
|
||||
return sid
|
||||
|
||||
|
||||
def end_session(sid: int, rx: int = 0, tx: int = 0) -> None:
|
||||
sessions = _load()
|
||||
for s in sessions:
|
||||
if s.get("id") == sid:
|
||||
start_dt = datetime.strptime(s["start"], "%Y-%m-%d %H:%M:%S")
|
||||
secs = int((datetime.now() - start_dt).total_seconds())
|
||||
s["end"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
s["duration"] = secs
|
||||
s["rx"] = rx
|
||||
s["tx"] = tx
|
||||
break
|
||||
_save(sessions)
|
||||
|
||||
|
||||
def get_sessions(n: int = 50) -> list[dict]:
|
||||
return list(reversed(_load()[-n:]))
|
||||
|
||||
|
||||
def clear() -> None:
|
||||
_save([])
|
||||
|
||||
|
||||
def fmt_duration(secs: int | None) -> str:
|
||||
if secs is None:
|
||||
return "—"
|
||||
h, r = divmod(int(secs), 3600)
|
||||
m, s = divmod(r, 60)
|
||||
if h:
|
||||
return f"{h}h {m:02d}m"
|
||||
if m:
|
||||
return f"{m}m {s:02d}s"
|
||||
return f"{s}s"
|
||||
|
||||
|
||||
def fmt_bytes(n: int) -> str:
|
||||
for unit, div in [("GB", 1_073_741_824), ("MB", 1_048_576), ("KB", 1024)]:
|
||||
if n >= div:
|
||||
return f"{n/div:.1f} {unit}"
|
||||
return f"{n} B"
|
||||
|
||||
|
||||
def _load() -> list[dict]:
|
||||
p = _path()
|
||||
if not os.path.exists(p):
|
||||
return []
|
||||
try:
|
||||
with open(p, "r", encoding="utf-8") as f:
|
||||
return json.load(f)
|
||||
except Exception:
|
||||
return []
|
||||
|
||||
|
||||
def _save(sessions: list[dict]) -> None:
|
||||
try:
|
||||
with open(_path(), "w", encoding="utf-8") as f:
|
||||
json.dump(sessions[-_MAX:], f, indent=2, ensure_ascii=False)
|
||||
except Exception:
|
||||
pass
|
||||
Reference in New Issue
Block a user