""" Metadata management for deb-mock """ import os import json import uuid from pathlib import Path from typing import Dict, Any, List, Optional from datetime import datetime from .exceptions import MetadataError class MetadataManager: """Manages build metadata capture and storage""" def __init__(self, config): self.config = config self.metadata_dir = Path(config.get_metadata_path()) self.metadata_dir.mkdir(parents=True, exist_ok=True) def store_metadata(self, metadata: Dict[str, Any]) -> str: """Store build metadata and return build ID""" # Generate unique build ID build_id = self._generate_build_id() # Add build ID to metadata metadata['build_id'] = build_id metadata['stored_at'] = datetime.now().isoformat() # Create metadata file metadata_file = self.metadata_dir / f"{build_id}.json" try: with open(metadata_file, 'w') as f: json.dump(metadata, f, indent=2, default=str) except Exception as e: raise MetadataError(f"Failed to store metadata: {e}") # Update build index self._update_build_index(build_id, metadata) return build_id def get_build_info(self, build_id: str) -> Optional[Dict[str, Any]]: """Get metadata for a specific build""" metadata_file = self.metadata_dir / f"{build_id}.json" if not metadata_file.exists(): return None try: with open(metadata_file, 'r') as f: return json.load(f) except Exception as e: raise MetadataError(f"Failed to load metadata for build {build_id}: {e}") def get_build_history(self, limit: int = None) -> List[Dict[str, Any]]: """Get build history, optionally limited to recent builds""" builds = [] # Load build index index_file = self.metadata_dir / "build_index.json" if not index_file.exists(): return builds try: with open(index_file, 'r') as f: build_index = json.load(f) except Exception as e: raise MetadataError(f"Failed to load build index: {e}") # Sort builds by timestamp (newest first) sorted_builds = sorted( build_index.values(), key=lambda x: x.get('timestamp', ''), reverse=True ) # Apply limit if specified if limit: sorted_builds = sorted_builds[:limit] # Load full metadata for each build for build_info in sorted_builds: build_id = build_info.get('build_id') if build_id: full_metadata = self.get_build_info(build_id) if full_metadata: builds.append(full_metadata) return builds def search_builds(self, criteria: Dict[str, Any]) -> List[Dict[str, Any]]: """Search builds based on criteria""" builds = [] all_builds = self.get_build_history() for build in all_builds: if self._matches_criteria(build, criteria): builds.append(build) return builds def delete_build_metadata(self, build_id: str) -> bool: """Delete metadata for a specific build""" metadata_file = self.metadata_dir / f"{build_id}.json" if not metadata_file.exists(): return False try: metadata_file.unlink() self._remove_from_index(build_id) return True except Exception as e: raise MetadataError(f"Failed to delete metadata for build {build_id}: {e}") def cleanup_old_metadata(self, days: int = 30) -> int: """Clean up metadata older than specified days""" cutoff_time = datetime.now().timestamp() - (days * 24 * 60 * 60) deleted_count = 0 all_builds = self.get_build_history() for build in all_builds: build_id = build.get('build_id') timestamp = build.get('timestamp') if timestamp: try: build_time = datetime.fromisoformat(timestamp).timestamp() if build_time < cutoff_time: if self.delete_build_metadata(build_id): deleted_count += 1 except ValueError: # Skip builds with invalid timestamps continue return deleted_count def export_metadata(self, build_id: str, format: str = 'json') -> str: """Export build metadata in specified format""" metadata = self.get_build_info(build_id) if not metadata: raise MetadataError(f"Build {build_id} not found") if format.lower() == 'json': return json.dumps(metadata, indent=2, default=str) elif format.lower() == 'yaml': import yaml return yaml.dump(metadata, default_flow_style=False) else: raise MetadataError(f"Unsupported export format: {format}") def _generate_build_id(self) -> str: """Generate a unique build ID""" return str(uuid.uuid4()) def _update_build_index(self, build_id: str, metadata: Dict[str, Any]) -> None: """Update the build index with new build information""" index_file = self.metadata_dir / "build_index.json" # Load existing index build_index = {} if index_file.exists(): try: with open(index_file, 'r') as f: build_index = json.load(f) except Exception: build_index = {} # Add new build to index build_index[build_id] = { 'build_id': build_id, 'source_package': metadata.get('source_package', ''), 'timestamp': metadata.get('timestamp', ''), 'build_success': metadata.get('build_success', False), 'package_name': metadata.get('build_metadata', {}).get('package_name', ''), 'package_version': metadata.get('build_metadata', {}).get('package_version', ''), 'architecture': metadata.get('build_metadata', {}).get('architecture', ''), 'suite': metadata.get('build_metadata', {}).get('suite', '') } # Save updated index try: with open(index_file, 'w') as f: json.dump(build_index, f, indent=2, default=str) except Exception as e: raise MetadataError(f"Failed to update build index: {e}") def _remove_from_index(self, build_id: str) -> None: """Remove a build from the index""" index_file = self.metadata_dir / "build_index.json" if not index_file.exists(): return try: with open(index_file, 'r') as f: build_index = json.load(f) except Exception: return if build_id in build_index: del build_index[build_id] try: with open(index_file, 'w') as f: json.dump(build_index, f, indent=2, default=str) except Exception as e: raise MetadataError(f"Failed to update build index: {e}") def _matches_criteria(self, build: Dict[str, Any], criteria: Dict[str, Any]) -> bool: """Check if a build matches the given criteria""" for key, value in criteria.items(): if key == 'package_name': build_package = build.get('build_metadata', {}).get('package_name', '') if value.lower() not in build_package.lower(): return False elif key == 'architecture': build_arch = build.get('build_metadata', {}).get('architecture', '') if value.lower() != build_arch.lower(): return False elif key == 'suite': build_suite = build.get('build_metadata', {}).get('suite', '') if value.lower() != build_suite.lower(): return False elif key == 'success': build_success = build.get('build_success', False) if value != build_success: return False elif key == 'date_after': build_timestamp = build.get('timestamp', '') if build_timestamp: try: build_time = datetime.fromisoformat(build_timestamp) criteria_time = datetime.fromisoformat(value) if build_time <= criteria_time: return False except ValueError: return False elif key == 'date_before': build_timestamp = build.get('timestamp', '') if build_timestamp: try: build_time = datetime.fromisoformat(build_timestamp) criteria_time = datetime.fromisoformat(value) if build_time >= criteria_time: return False except ValueError: return False return True