Build Proxmox visualizer app
This commit is contained in:
@@ -0,0 +1,33 @@
|
||||
import re
|
||||
from typing import Optional
|
||||
|
||||
MARKER_START = "--- [prox-visualizer-metadata] ---"
|
||||
|
||||
def extract_duplicate_id(description: Optional[str]) -> Optional[int]:
|
||||
"""Extrait l'ID de doublon caché dans la note Proxmox via Regex."""
|
||||
if not description:
|
||||
return None
|
||||
|
||||
# Cherche la clé duplicate_id=chiffres
|
||||
match = re.search(r'duplicate_id=(\d+)', description)
|
||||
if match:
|
||||
return int(match.group(1))
|
||||
return None
|
||||
|
||||
def inject_duplicate_id(description: Optional[str], duplicate_id: int) -> str:
|
||||
"""
|
||||
Insère ou met à jour l'ID de doublon à la fin de la note existante,
|
||||
sans effacer les notes déjà écrites par l'utilisateur.
|
||||
"""
|
||||
clean_desc = ""
|
||||
if description:
|
||||
# Si un ancien bloc de métadonnées existe, on le retire pour repartir sur du propre
|
||||
if MARKER_START in description:
|
||||
clean_desc = description.split(MARKER_START)[0].strip()
|
||||
else:
|
||||
clean_desc = description.strip()
|
||||
|
||||
# Génération du nouveau bloc de métadonnées
|
||||
metadata_block = f"\n\n{MARKER_START}\nduplicate_id={duplicate_id}"
|
||||
|
||||
return f"{clean_desc}{metadata_block}".strip()
|
||||
@@ -0,0 +1,163 @@
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
import httpx
|
||||
|
||||
from app.core.config import settings
|
||||
from app.models.schemas import ProxmoxServerConfig, ServerScanStatus, VmRecord
|
||||
from app.services.metadata import extract_duplicate_id
|
||||
|
||||
|
||||
class ProxmoxClient:
|
||||
def __init__(self, server: ProxmoxServerConfig):
|
||||
self.server = server
|
||||
self.base_url = str(server.url).rstrip("/")
|
||||
self.errors: List[str] = []
|
||||
|
||||
async def __aenter__(self) -> "ProxmoxClient":
|
||||
self.client = httpx.AsyncClient(
|
||||
base_url=f"{self.base_url}/api2/json",
|
||||
headers={
|
||||
"Authorization": (
|
||||
f"PVEAPIToken={self.server.token_name}={self.server.token_value}"
|
||||
)
|
||||
},
|
||||
timeout=settings.PROXMOX_TIMEOUT,
|
||||
verify=settings.PROXMOX_VERIFY_SSL,
|
||||
)
|
||||
return self
|
||||
|
||||
async def __aexit__(self, exc_type, exc, tb) -> None:
|
||||
await self.client.aclose()
|
||||
|
||||
async def scan(self) -> tuple[ServerScanStatus, List[VmRecord]]:
|
||||
items = await self._scan_nodes()
|
||||
if not items:
|
||||
fallback_items = await self._scan_cluster_resources()
|
||||
items.extend(fallback_items)
|
||||
if not items and await self._has_no_effective_permissions():
|
||||
self.errors.append(
|
||||
"permissions effectives vides pour ce token; Proxmox filtre probablement les VM/LXC"
|
||||
)
|
||||
|
||||
status = ServerScanStatus(
|
||||
name=self.server.name,
|
||||
url=self.base_url,
|
||||
ok=not self.errors or bool(items),
|
||||
vm_count=len(items),
|
||||
errors=self.errors,
|
||||
)
|
||||
return status, items
|
||||
|
||||
async def _has_no_effective_permissions(self) -> bool:
|
||||
permissions = await self._get_data("/access/permissions?path=/", optional=True)
|
||||
if not isinstance(permissions, dict):
|
||||
return False
|
||||
if not permissions:
|
||||
return True
|
||||
return all(not value for value in permissions.values())
|
||||
|
||||
async def _get_data(self, path: str, *, optional: bool = False) -> Optional[Any]:
|
||||
try:
|
||||
response = await self.client.get(path)
|
||||
response.raise_for_status()
|
||||
payload = response.json()
|
||||
except httpx.HTTPStatusError as exc:
|
||||
message = self._format_http_error(path, exc)
|
||||
if optional and exc.response.status_code in {401, 403, 404}:
|
||||
return None
|
||||
self.errors.append(message)
|
||||
return None
|
||||
except httpx.HTTPError as exc:
|
||||
self.errors.append(f"{path}: {exc}")
|
||||
return None
|
||||
except ValueError:
|
||||
self.errors.append(f"{path}: réponse JSON invalide")
|
||||
return None
|
||||
|
||||
return payload.get("data")
|
||||
|
||||
def _format_http_error(self, path: str, exc: httpx.HTTPStatusError) -> str:
|
||||
detail = exc.response.text.strip()
|
||||
if len(detail) > 160:
|
||||
detail = f"{detail[:157]}..."
|
||||
return f"{path}: HTTP {exc.response.status_code} {detail}".strip()
|
||||
|
||||
async def _scan_nodes(self) -> List[VmRecord]:
|
||||
nodes = await self._get_data("/nodes")
|
||||
if not isinstance(nodes, list):
|
||||
return []
|
||||
|
||||
items: List[VmRecord] = []
|
||||
for node in nodes:
|
||||
node_name = node.get("node")
|
||||
if not node_name:
|
||||
continue
|
||||
items.extend(await self._scan_node_vms(node_name, "qemu"))
|
||||
items.extend(await self._scan_node_vms(node_name, "lxc"))
|
||||
return items
|
||||
|
||||
async def _scan_node_vms(self, node: str, vm_type: str) -> List[VmRecord]:
|
||||
path = f"/nodes/{node}/{vm_type}"
|
||||
records = await self._get_data(path)
|
||||
if not isinstance(records, list):
|
||||
return []
|
||||
|
||||
items = []
|
||||
for record in records:
|
||||
vmid = record.get("vmid")
|
||||
if vmid is None:
|
||||
continue
|
||||
config = await self._get_data(
|
||||
f"{path}/{vmid}/config",
|
||||
optional=True,
|
||||
)
|
||||
items.append(self._to_vm_record(node, vm_type, record, config))
|
||||
return items
|
||||
|
||||
async def _scan_cluster_resources(self) -> List[VmRecord]:
|
||||
resources = await self._get_data("/cluster/resources?type=vm")
|
||||
if not isinstance(resources, list):
|
||||
return []
|
||||
|
||||
items = []
|
||||
for resource in resources:
|
||||
vm_type = resource.get("type")
|
||||
if vm_type not in {"qemu", "lxc"}:
|
||||
continue
|
||||
node = resource.get("node") or "unknown"
|
||||
items.append(self._to_vm_record(node, vm_type, resource, None))
|
||||
return items
|
||||
|
||||
def _to_vm_record(
|
||||
self,
|
||||
node: str,
|
||||
vm_type: str,
|
||||
record: Dict[str, Any],
|
||||
config: Optional[Dict[str, Any]],
|
||||
) -> VmRecord:
|
||||
vmid = int(record["vmid"])
|
||||
description = None
|
||||
tags = record.get("tags")
|
||||
if isinstance(config, dict):
|
||||
description = config.get("description")
|
||||
tags = config.get("tags", tags)
|
||||
|
||||
return VmRecord(
|
||||
id=f"{self.server.name}:{node}:{vm_type}:{vmid}",
|
||||
server=self.server.name,
|
||||
node=node,
|
||||
vmid=vmid,
|
||||
type=vm_type,
|
||||
name=record.get("name") or record.get("hostname") or f"{vm_type}-{vmid}",
|
||||
status=record.get("status"),
|
||||
cpu=record.get("cpu"),
|
||||
mem=record.get("mem"),
|
||||
maxmem=record.get("maxmem"),
|
||||
disk=record.get("disk"),
|
||||
maxdisk=record.get("maxdisk"),
|
||||
uptime=record.get("uptime"),
|
||||
tags=tags,
|
||||
duplicate_id=extract_duplicate_id(description),
|
||||
description_available=description is not None,
|
||||
raw=record,
|
||||
)
|
||||
@@ -0,0 +1,128 @@
|
||||
import asyncio
|
||||
import re
|
||||
from collections import defaultdict
|
||||
from typing import Dict, Iterable, List, Tuple
|
||||
|
||||
from pydantic import ValidationError
|
||||
|
||||
from app.models.schemas import (
|
||||
DuplicateGroup,
|
||||
ProxmoxServerConfig,
|
||||
ScanResponse,
|
||||
ServerScanStatus,
|
||||
VmRecord,
|
||||
)
|
||||
from app.services.proxmox_api import ProxmoxClient
|
||||
|
||||
|
||||
def parse_server_configs(raw_servers: Iterable[dict]) -> Tuple[List[ProxmoxServerConfig], List[str]]:
|
||||
servers: List[ProxmoxServerConfig] = []
|
||||
errors: List[str] = []
|
||||
|
||||
for index, raw_server in enumerate(raw_servers, start=1):
|
||||
try:
|
||||
servers.append(ProxmoxServerConfig.model_validate(raw_server))
|
||||
except ValidationError as exc:
|
||||
errors.append(f"serveur #{index}: {exc.errors()[0]['msg']}")
|
||||
|
||||
return servers, errors
|
||||
|
||||
|
||||
async def scan_servers(raw_servers: Iterable[dict]) -> ScanResponse:
|
||||
servers, config_errors = parse_server_configs(raw_servers)
|
||||
if not servers:
|
||||
return ScanResponse(
|
||||
servers=[
|
||||
ServerScanStatus(
|
||||
name="configuration",
|
||||
url="",
|
||||
ok=False,
|
||||
errors=config_errors or ["aucun serveur Proxmox configure"],
|
||||
)
|
||||
],
|
||||
items=[],
|
||||
duplicates=[],
|
||||
)
|
||||
|
||||
results = await asyncio.gather(*[_scan_one(server) for server in servers])
|
||||
statuses: List[ServerScanStatus] = []
|
||||
items: List[VmRecord] = []
|
||||
|
||||
if config_errors:
|
||||
statuses.append(
|
||||
ServerScanStatus(
|
||||
name="configuration",
|
||||
url="",
|
||||
ok=False,
|
||||
errors=config_errors,
|
||||
)
|
||||
)
|
||||
|
||||
for status, server_items in results:
|
||||
statuses.append(status)
|
||||
items.extend(server_items)
|
||||
|
||||
return ScanResponse(
|
||||
servers=statuses,
|
||||
items=sorted(items, key=lambda vm: (vm.server, vm.node, vm.vmid)),
|
||||
duplicates=find_duplicates(items),
|
||||
)
|
||||
|
||||
|
||||
async def _scan_one(server: ProxmoxServerConfig) -> tuple[ServerScanStatus, List[VmRecord]]:
|
||||
async with ProxmoxClient(server) as client:
|
||||
return await client.scan()
|
||||
|
||||
|
||||
def find_duplicates(items: List[VmRecord]) -> List[DuplicateGroup]:
|
||||
groups: List[DuplicateGroup] = []
|
||||
seen_item_sets = set()
|
||||
|
||||
by_metadata: Dict[int, List[VmRecord]] = defaultdict(list)
|
||||
for item in items:
|
||||
if item.duplicate_id is not None:
|
||||
by_metadata[item.duplicate_id].append(item)
|
||||
|
||||
for duplicate_id, members in sorted(by_metadata.items()):
|
||||
if len(members) < 2:
|
||||
continue
|
||||
groups.append(
|
||||
DuplicateGroup(
|
||||
id=f"metadata:{duplicate_id}",
|
||||
label=f"duplicate_id={duplicate_id}",
|
||||
reason="metadata",
|
||||
count=len(members),
|
||||
items=members,
|
||||
)
|
||||
)
|
||||
seen_item_sets.add(frozenset(item.id for item in members))
|
||||
|
||||
by_name: Dict[str, List[VmRecord]] = defaultdict(list)
|
||||
for item in items:
|
||||
normalized = normalize_name(item.name)
|
||||
if normalized:
|
||||
by_name[normalized].append(item)
|
||||
|
||||
for normalized, members in sorted(by_name.items()):
|
||||
if len(members) < 2:
|
||||
continue
|
||||
item_set = frozenset(item.id for item in members)
|
||||
if item_set in seen_item_sets:
|
||||
continue
|
||||
groups.append(
|
||||
DuplicateGroup(
|
||||
id=f"name:{normalized}",
|
||||
label=members[0].name,
|
||||
reason="name",
|
||||
count=len(members),
|
||||
items=members,
|
||||
)
|
||||
)
|
||||
|
||||
return groups
|
||||
|
||||
|
||||
def normalize_name(name: str) -> str:
|
||||
normalized = name.strip().lower()
|
||||
normalized = re.sub(r"\s+", " ", normalized)
|
||||
return normalized
|
||||
Reference in New Issue
Block a user