""" Metadata management for deb-mock """ import json import uuid from datetime import datetime from pathlib import Path from typing import Any, Dict, List, Optional 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()) # Ensure the metadata directory exists try: self.metadata_dir.mkdir(parents=True, exist_ok=True) except Exception as e: # If we can't create the directory, use a fallback import tempfile self.metadata_dir = Path(tempfile.gettempdir()) / "deb-mock-metadata" 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