309 lines
11 KiB
Python
309 lines
11 KiB
Python
"""
|
|
Dell Repository Manager (DRM) 연동 모듈
|
|
backend/services/drm_catalog_sync.py
|
|
|
|
Dell Repository Manager에서 생성한 로컬 리포지토리 카탈로그를 파싱하여
|
|
펌웨어 정보를 가져오는 기능
|
|
"""
|
|
|
|
import os
|
|
import xml.etree.ElementTree as ET
|
|
from pathlib import Path
|
|
from typing import List, Dict, Optional
|
|
from backend.models.firmware_version import FirmwareVersion, db
|
|
from datetime import datetime
|
|
|
|
|
|
class DRMCatalogSync:
|
|
"""Dell Repository Manager 카탈로그 동기화"""
|
|
|
|
def __init__(self, drm_repository_path: str = None):
|
|
"""
|
|
Args:
|
|
drm_repository_path: DRM 리포지토리 경로
|
|
예: C:/Dell/Repository 또는 네트워크 경로
|
|
"""
|
|
self.repository_path = drm_repository_path
|
|
self.catalog_file = None
|
|
|
|
if drm_repository_path:
|
|
# DRM은 보통 catalog 폴더에 XML 파일 생성
|
|
possible_catalogs = [
|
|
os.path.join(drm_repository_path, 'catalog', 'Catalog.xml'),
|
|
os.path.join(drm_repository_path, 'Catalog.xml'),
|
|
os.path.join(drm_repository_path, 'catalog.xml'),
|
|
]
|
|
|
|
for catalog_path in possible_catalogs:
|
|
if os.path.exists(catalog_path):
|
|
self.catalog_file = catalog_path
|
|
print(f"[INFO] DRM 카탈로그 파일 발견: {catalog_path}")
|
|
break
|
|
|
|
def sync_from_drm_repository(self, model: str = "PowerEdge R750") -> int:
|
|
"""
|
|
DRM 리포지토리에서 펌웨어 정보 동기화
|
|
|
|
Args:
|
|
model: 서버 모델명
|
|
|
|
Returns:
|
|
동기화된 펌웨어 개수
|
|
"""
|
|
if not self.catalog_file or not os.path.exists(self.catalog_file):
|
|
print(f"[오류] DRM 카탈로그 파일을 찾을 수 없습니다: {self.repository_path}")
|
|
return 0
|
|
|
|
try:
|
|
print(f"[INFO] DRM 카탈로그 파싱 중: {self.catalog_file}")
|
|
|
|
# XML 파싱
|
|
tree = ET.parse(self.catalog_file)
|
|
root = tree.getroot()
|
|
|
|
count = 0
|
|
|
|
# DRM 카탈로그 구조 파싱
|
|
for pkg in root.findall(".//SoftwareComponent"):
|
|
# 모델 필터링
|
|
supported_models = self._get_supported_models(pkg)
|
|
|
|
if model and model not in supported_models:
|
|
continue
|
|
|
|
# 펌웨어 정보 추출
|
|
firmware_info = self._extract_firmware_info(pkg, model)
|
|
|
|
if firmware_info:
|
|
# DB에 저장
|
|
if self._save_firmware_to_db(firmware_info):
|
|
count += 1
|
|
|
|
print(f"✓ DRM에서 {model} 관련 펌웨어 {count}개 동기화 완료")
|
|
return count
|
|
|
|
except Exception as e:
|
|
print(f"[오류] DRM 카탈로그 동기화 실패: {str(e)}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return 0
|
|
|
|
def _get_supported_models(self, pkg_element) -> List[str]:
|
|
"""패키지가 지원하는 서버 모델 목록 추출"""
|
|
models = []
|
|
|
|
# Dell 카탈로그 XML 구조
|
|
for display in pkg_element.findall(".//SupportedSystems/Brand/Model/Display"):
|
|
if display.text:
|
|
models.append(display.text.strip())
|
|
|
|
return models
|
|
|
|
def _extract_firmware_info(self, pkg_element, model: str) -> Optional[Dict]:
|
|
"""패키지에서 펌웨어 정보 추출"""
|
|
try:
|
|
name = pkg_element.findtext("Name")
|
|
version = pkg_element.findtext("vendorVersion") # DRM은 vendorVersion 사용
|
|
if not version:
|
|
version = pkg_element.findtext("dellVersion") # 또는 dellVersion
|
|
|
|
release_date = pkg_element.findtext("dateTime")
|
|
path = pkg_element.findtext("path")
|
|
category = pkg_element.findtext("Category")
|
|
|
|
# 중요도 판단
|
|
importance = pkg_element.findtext("ImportanceDisplay")
|
|
is_critical = importance and "Critical" in importance
|
|
|
|
# 파일 정보
|
|
file_name = None
|
|
file_size_mb = None
|
|
|
|
if path:
|
|
file_name = os.path.basename(path)
|
|
# 실제 파일이 있으면 크기 확인
|
|
if self.repository_path:
|
|
full_path = os.path.join(self.repository_path, path)
|
|
if os.path.exists(full_path):
|
|
file_size_mb = os.path.getsize(full_path) // (1024 * 1024)
|
|
|
|
if not name or not version:
|
|
return None
|
|
|
|
return {
|
|
'component_name': name,
|
|
'latest_version': version,
|
|
'server_model': model,
|
|
'vendor': 'Dell',
|
|
'release_date': self._parse_date(release_date),
|
|
'download_url': f"https://downloads.dell.com/{path}" if path else None,
|
|
'file_name': file_name,
|
|
'file_size_mb': file_size_mb,
|
|
'component_type': category,
|
|
'is_critical': is_critical,
|
|
'notes': f"DRM에서 동기화 - {category}" if category else "DRM에서 동기화"
|
|
}
|
|
|
|
except Exception as e:
|
|
print(f"[경고] 펌웨어 정보 추출 실패: {str(e)}")
|
|
return None
|
|
|
|
def _parse_date(self, date_str: str) -> Optional[str]:
|
|
"""날짜 문자열 파싱"""
|
|
if not date_str:
|
|
return None
|
|
|
|
try:
|
|
# DRM 날짜 형식: 2024-03-15T10:30:00
|
|
dt = datetime.fromisoformat(date_str.replace('Z', '+00:00'))
|
|
return dt.strftime('%Y-%m-%d')
|
|
except:
|
|
return date_str[:10] if len(date_str) >= 10 else None
|
|
|
|
def _save_firmware_to_db(self, firmware_info: Dict) -> bool:
|
|
"""펌웨어 정보를 DB에 저장"""
|
|
try:
|
|
# 중복 확인
|
|
existing = FirmwareVersion.query.filter_by(
|
|
component_name=firmware_info['component_name'],
|
|
latest_version=firmware_info['latest_version'],
|
|
server_model=firmware_info['server_model']
|
|
).first()
|
|
|
|
if existing:
|
|
# 이미 존재하면 업데이트
|
|
for key, value in firmware_info.items():
|
|
if hasattr(existing, key) and value is not None:
|
|
setattr(existing, key, value)
|
|
db.session.commit()
|
|
return False # 새로 추가된 것은 아님
|
|
else:
|
|
# 새로 추가
|
|
version = FirmwareVersion(**firmware_info)
|
|
db.session.add(version)
|
|
db.session.commit()
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"[오류] DB 저장 실패: {str(e)}")
|
|
db.session.rollback()
|
|
return False
|
|
|
|
def list_available_models(self) -> List[str]:
|
|
"""DRM 리포지토리에서 사용 가능한 모델 목록 조회"""
|
|
if not self.catalog_file or not os.path.exists(self.catalog_file):
|
|
return []
|
|
|
|
try:
|
|
tree = ET.parse(self.catalog_file)
|
|
root = tree.getroot()
|
|
|
|
models = set()
|
|
|
|
for display in root.findall(".//SupportedSystems/Brand/Model/Display"):
|
|
if display.text:
|
|
models.add(display.text.strip())
|
|
|
|
return sorted(list(models))
|
|
|
|
except Exception as e:
|
|
print(f"[오류] 모델 목록 조회 실패: {str(e)}")
|
|
return []
|
|
|
|
def get_repository_info(self) -> Dict:
|
|
"""DRM 리포지토리 정보 조회"""
|
|
if not self.catalog_file or not os.path.exists(self.catalog_file):
|
|
return {
|
|
'exists': False,
|
|
'path': self.repository_path,
|
|
'catalog_file': None
|
|
}
|
|
|
|
try:
|
|
tree = ET.parse(self.catalog_file)
|
|
root = tree.getroot()
|
|
|
|
# 카탈로그 메타데이터
|
|
base_location = root.findtext(".//BaseLocation")
|
|
release_id = root.findtext(".//ReleaseID")
|
|
|
|
# 패키지 수 계산
|
|
total_packages = len(root.findall(".//SoftwareComponent"))
|
|
|
|
return {
|
|
'exists': True,
|
|
'path': self.repository_path,
|
|
'catalog_file': self.catalog_file,
|
|
'base_location': base_location,
|
|
'release_id': release_id,
|
|
'total_packages': total_packages,
|
|
'available_models': self.list_available_models()
|
|
}
|
|
|
|
except Exception as e:
|
|
return {
|
|
'exists': True,
|
|
'path': self.repository_path,
|
|
'catalog_file': self.catalog_file,
|
|
'error': str(e)
|
|
}
|
|
|
|
|
|
# ========================================
|
|
# 편의 함수
|
|
# ========================================
|
|
|
|
def sync_from_drm(repository_path: str, model: str = "PowerEdge R750") -> Dict:
|
|
"""
|
|
DRM 리포지토리에서 펌웨어 동기화 (간편 함수)
|
|
|
|
Args:
|
|
repository_path: DRM 리포지토리 경로
|
|
model: 서버 모델명
|
|
|
|
Returns:
|
|
{'success': bool, 'count': int, 'message': str}
|
|
"""
|
|
try:
|
|
drm = DRMCatalogSync(repository_path)
|
|
|
|
# 리포지토리 확인
|
|
info = drm.get_repository_info()
|
|
if not info['exists']:
|
|
return {
|
|
'success': False,
|
|
'count': 0,
|
|
'message': f'DRM 리포지토리를 찾을 수 없습니다: {repository_path}'
|
|
}
|
|
|
|
# 동기화
|
|
count = drm.sync_from_drm_repository(model)
|
|
|
|
return {
|
|
'success': True,
|
|
'count': count,
|
|
'message': f'{model} 펌웨어 {count}개 동기화 완료',
|
|
'repository_info': info
|
|
}
|
|
|
|
except Exception as e:
|
|
return {
|
|
'success': False,
|
|
'count': 0,
|
|
'message': f'DRM 동기화 실패: {str(e)}'
|
|
}
|
|
|
|
|
|
def check_drm_repository(repository_path: str) -> Dict:
|
|
"""
|
|
DRM 리포지토리 상태 확인
|
|
|
|
Args:
|
|
repository_path: DRM 리포지토리 경로
|
|
|
|
Returns:
|
|
리포지토리 정보
|
|
"""
|
|
drm = DRMCatalogSync(repository_path)
|
|
return drm.get_repository_info()
|