118 lines
3.3 KiB
Python
118 lines
3.3 KiB
Python
"""Proxmox Backup Server REST API Client — Backup-Status und Datastore-Info."""
|
|
|
|
import requests
|
|
import urllib3
|
|
from datetime import datetime
|
|
from typing import Optional
|
|
|
|
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
|
|
|
PBS_URL = ""
|
|
PBS_USER = "root@pam"
|
|
PBS_PASS = ""
|
|
_ticket_cache = ""
|
|
|
|
|
|
def init(cfg):
|
|
global PBS_URL, PBS_PASS, _ticket_cache
|
|
_ticket_cache = ""
|
|
pbs_ip = cfg.raw.get("SRV_PBS_MU", "100.99.139.22")
|
|
PBS_URL = f"https://{pbs_ip}:8007/api2/json"
|
|
PBS_PASS = cfg.passwords.get("default", "")
|
|
|
|
|
|
def _get_ticket() -> str:
|
|
global _ticket_cache
|
|
if _ticket_cache:
|
|
return _ticket_cache
|
|
try:
|
|
r = requests.post(
|
|
f"{PBS_URL}/access/ticket",
|
|
data={"username": PBS_USER, "password": PBS_PASS},
|
|
verify=False, timeout=5,
|
|
)
|
|
r.raise_for_status()
|
|
_ticket_cache = r.json()["data"]["ticket"]
|
|
return _ticket_cache
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def _get(endpoint: str) -> Optional[dict | list]:
|
|
ticket = _get_ticket()
|
|
if not ticket:
|
|
return None
|
|
try:
|
|
r = requests.get(
|
|
f"{PBS_URL}{endpoint}",
|
|
cookies={"PBSAuthCookie": ticket},
|
|
verify=False, timeout=10,
|
|
)
|
|
r.raise_for_status()
|
|
return r.json().get("data")
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def get_datastore_usage() -> list[dict]:
|
|
data = _get("/status/datastore-usage")
|
|
if not data:
|
|
return []
|
|
results = []
|
|
for d in data:
|
|
total = d.get("total", 0) / 1024**3
|
|
used = d.get("used", 0) / 1024**3
|
|
avail = d.get("avail", 0) / 1024**3
|
|
pct = (used / total * 100) if total > 0 else 0
|
|
results.append({
|
|
"store": d.get("store", "?"),
|
|
"total_gb": total,
|
|
"used_gb": used,
|
|
"avail_gb": avail,
|
|
"used_pct": pct,
|
|
})
|
|
return results
|
|
|
|
|
|
def get_recent_snapshots(datastore: str = "nvme-pool", limit: int = 10) -> list[dict]:
|
|
data = _get(f"/admin/datastore/{datastore}/snapshots")
|
|
if not data:
|
|
return []
|
|
data.sort(key=lambda s: s.get("backup-time", 0), reverse=True)
|
|
results = []
|
|
for s in data[:limit]:
|
|
ts = datetime.fromtimestamp(s.get("backup-time", 0))
|
|
results.append({
|
|
"id": s.get("backup-id", "?"),
|
|
"type": s.get("backup-type", "?"),
|
|
"time": ts.strftime("%Y-%m-%d %H:%M"),
|
|
"size_gb": s.get("size", 0) / 1024**3,
|
|
})
|
|
return results
|
|
|
|
|
|
def get_snapshot_count(datastore: str = "nvme-pool") -> int:
|
|
data = _get(f"/admin/datastore/{datastore}/snapshots")
|
|
return len(data) if data else 0
|
|
|
|
|
|
def format_overview() -> str:
|
|
stores = get_datastore_usage()
|
|
if not stores:
|
|
return "PBS nicht erreichbar."
|
|
|
|
lines = ["PBS Muldenstein — Backup-Status\n"]
|
|
lines.append("Datastores:")
|
|
for s in stores:
|
|
if s["total_gb"] < 1:
|
|
continue
|
|
lines.append(f" {s['store']}: {s['used_gb']:.0f}/{s['total_gb']:.0f} GB ({s['used_pct']:.0f}%)")
|
|
|
|
snaps = get_recent_snapshots(limit=5)
|
|
total_snaps = get_snapshot_count()
|
|
if snaps:
|
|
lines.append(f"\nLetzte Backups ({total_snaps} total):")
|
|
for s in snaps:
|
|
lines.append(f" {s['time']} CT {s['id']} ({s['size_gb']:.1f} GB)")
|
|
|
|
return "\n".join(lines)
|