사이드바 브랜딩을 PharmQ Super Admin (PSA)로 업데이트
- base.html 상단 네비게이션 브랜딩 변경 - 팜큐 약국 관리 시스템 → PharmQ Super Admin (PSA) - UI 일관성 향상을 위한 브랜딩 통합 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
165
farmq-admin/utils/proxmox_client.py
Normal file
165
farmq-admin/utils/proxmox_client.py
Normal file
@@ -0,0 +1,165 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Proxmox VE API 클라이언트
|
||||
"""
|
||||
|
||||
import requests
|
||||
import json
|
||||
import urllib3
|
||||
from urllib.parse import quote_plus
|
||||
from typing import Dict, List, Optional, Tuple
|
||||
|
||||
# SSL 경고 무시
|
||||
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
||||
|
||||
class ProxmoxClient:
|
||||
def __init__(self, host: str, username: str = "root@pam", password: str = "", api_token: str = ""):
|
||||
self.host = host
|
||||
self.username = username
|
||||
self.password = password
|
||||
self.api_token = api_token
|
||||
self.base_url = f"https://{host}:443/api2/json"
|
||||
self.session = requests.Session()
|
||||
self.session.verify = False
|
||||
self.ticket = None
|
||||
self.csrf_token = None
|
||||
|
||||
def login(self) -> bool:
|
||||
"""세션 쿠키 방식으로 로그인"""
|
||||
if self.api_token:
|
||||
# API Token 방식
|
||||
self.session.headers.update({
|
||||
'Authorization': f'PVEAPIToken={self.api_token}'
|
||||
})
|
||||
return self._test_connection()
|
||||
else:
|
||||
# 패스워드 방식
|
||||
login_data = {
|
||||
'username': self.username,
|
||||
'password': self.password
|
||||
}
|
||||
|
||||
try:
|
||||
response = self.session.post(
|
||||
f"{self.base_url}/access/ticket",
|
||||
data=login_data,
|
||||
timeout=10
|
||||
)
|
||||
|
||||
if response.status_code == 200:
|
||||
data = response.json()['data']
|
||||
self.ticket = data['ticket']
|
||||
self.csrf_token = data['CSRFPreventionToken']
|
||||
|
||||
# 쿠키와 헤더 설정
|
||||
self.session.cookies.update({'PVEAuthCookie': self.ticket})
|
||||
self.session.headers.update({'CSRFPreventionToken': self.csrf_token})
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"로그인 실패: {e}")
|
||||
return False
|
||||
|
||||
return False
|
||||
|
||||
def _test_connection(self) -> bool:
|
||||
"""연결 테스트"""
|
||||
try:
|
||||
response = self.session.get(f"{self.base_url}/version", timeout=10)
|
||||
return response.status_code == 200
|
||||
except:
|
||||
return False
|
||||
|
||||
def get_vm_list(self) -> List[Dict]:
|
||||
"""VM 목록 조회"""
|
||||
try:
|
||||
response = self.session.get(
|
||||
f"{self.base_url}/cluster/resources?type=vm",
|
||||
timeout=10
|
||||
)
|
||||
|
||||
if response.status_code == 200:
|
||||
return response.json()['data']
|
||||
|
||||
except Exception as e:
|
||||
print(f"VM 목록 조회 실패: {e}")
|
||||
|
||||
return []
|
||||
|
||||
def get_vm_status(self, node: str, vmid: int) -> Dict:
|
||||
"""특정 VM 상태 확인"""
|
||||
try:
|
||||
response = self.session.get(
|
||||
f"{self.base_url}/nodes/{node}/qemu/{vmid}/status/current",
|
||||
timeout=10
|
||||
)
|
||||
|
||||
if response.status_code == 200:
|
||||
return response.json()['data']
|
||||
|
||||
except Exception as e:
|
||||
print(f"VM {vmid} 상태 조회 실패: {e}")
|
||||
|
||||
return {}
|
||||
|
||||
def get_vnc_ticket(self, node: str, vmid: int) -> Optional[Dict]:
|
||||
"""VNC 접속 티켓 생성"""
|
||||
try:
|
||||
data = {'websocket': '1'}
|
||||
response = self.session.post(
|
||||
f"{self.base_url}/nodes/{node}/qemu/{vmid}/vncproxy",
|
||||
data=data,
|
||||
timeout=10
|
||||
)
|
||||
|
||||
if response.status_code == 200:
|
||||
vnc_data = response.json()['data']
|
||||
|
||||
# WebSocket URL 생성
|
||||
encoded_ticket = quote_plus(vnc_data['ticket'])
|
||||
vnc_data['websocket_url'] = f"wss://{self.host}:443/api2/json/nodes/{node}/qemu/{vmid}/vncwebsocket?port={vnc_data['port']}&vncticket={encoded_ticket}"
|
||||
|
||||
return vnc_data
|
||||
|
||||
except Exception as e:
|
||||
print(f"VNC 티켓 생성 실패: {e}")
|
||||
|
||||
return None
|
||||
|
||||
def start_vm(self, node: str, vmid: int) -> bool:
|
||||
"""VM 시작"""
|
||||
try:
|
||||
response = self.session.post(
|
||||
f"{self.base_url}/nodes/{node}/qemu/{vmid}/status/start",
|
||||
timeout=30
|
||||
)
|
||||
return response.status_code == 200
|
||||
|
||||
except Exception as e:
|
||||
print(f"VM {vmid} 시작 실패: {e}")
|
||||
return False
|
||||
|
||||
def stop_vm(self, node: str, vmid: int) -> bool:
|
||||
"""VM 정지"""
|
||||
try:
|
||||
response = self.session.post(
|
||||
f"{self.base_url}/nodes/{node}/qemu/{vmid}/status/stop",
|
||||
timeout=30
|
||||
)
|
||||
return response.status_code == 200
|
||||
|
||||
except Exception as e:
|
||||
print(f"VM {vmid} 정지 실패: {e}")
|
||||
return False
|
||||
|
||||
def get_nodes(self) -> List[Dict]:
|
||||
"""노드 목록 조회"""
|
||||
try:
|
||||
response = self.session.get(f"{self.base_url}/nodes", timeout=10)
|
||||
if response.status_code == 200:
|
||||
return response.json()['data']
|
||||
except Exception as e:
|
||||
print(f"노드 목록 조회 실패: {e}")
|
||||
|
||||
return []
|
||||
Reference in New Issue
Block a user