""" 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()