#!/usr/bin/env python3 """ Composer Build History for Debian Forge This module provides build history tracking, storage, and retrieval for composer-based builds. """ import json import sqlite3 import hashlib from typing import Dict, List, Optional, Any from dataclasses import dataclass, asdict from datetime import datetime, timedelta from pathlib import Path import threading @dataclass class BuildRecord: """Represents a complete build record""" build_id: str blueprint: str target: str architecture: str status: str created_at: datetime completed_at: Optional[datetime] duration: Optional[float] # in seconds metadata: Dict[str, Any] logs: List[str] artifacts: List[str] error_message: Optional[str] class BuildHistoryDB: """SQLite-based build history database""" def __init__(self, db_path: str = "build_history.db"): self.db_path = db_path self.lock = threading.Lock() self._init_database() def _init_database(self): """Initialize the database schema""" with self.lock: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() # Create builds table cursor.execute(''' CREATE TABLE IF NOT EXISTS builds ( build_id TEXT PRIMARY KEY, blueprint TEXT NOT NULL, target TEXT NOT NULL, architecture TEXT NOT NULL, status TEXT NOT NULL, created_at TEXT NOT NULL, completed_at TEXT, duration REAL, metadata TEXT, logs TEXT, artifacts TEXT, error_message TEXT ) ''') # Create indexes for common queries cursor.execute('CREATE INDEX IF NOT EXISTS idx_blueprint ON builds(blueprint)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_status ON builds(status)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_created_at ON builds(created_at)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_architecture ON builds(architecture)') conn.commit() conn.close() def add_build(self, build_record: BuildRecord) -> bool: """Add a new build record""" try: with self.lock: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(''' INSERT OR REPLACE INTO builds (build_id, blueprint, target, architecture, status, created_at, completed_at, duration, metadata, logs, artifacts, error_message) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ''', ( build_record.build_id, build_record.blueprint, build_record.target, build_record.architecture, build_record.status, build_record.created_at.isoformat(), build_record.completed_at.isoformat() if build_record.completed_at else None, build_record.duration, json.dumps(build_record.metadata), json.dumps(build_record.logs), json.dumps(build_record.artifacts), build_record.error_message )) conn.commit() conn.close() return True except Exception as e: print(f"Failed to add build record: {e}") return False def update_build_status(self, build_id: str, status: str, **kwargs) -> bool: """Update build status and other fields""" try: with self.lock: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() # Build update query dynamically update_fields = [] values = [] if 'status' in kwargs: update_fields.append('status = ?') values.append(kwargs['status']) if 'completed_at' in kwargs: update_fields.append('completed_at = ?') values.append(kwargs['completed_at'].isoformat()) if 'duration' in kwargs: update_fields.append('duration = ?') values.append(kwargs['duration']) if 'logs' in kwargs: update_fields.append('logs = ?') values.append(json.dumps(kwargs['logs'])) if 'artifacts' in kwargs: update_fields.append('artifacts = ?') values.append(json.dumps(kwargs['artifacts'])) if 'error_message' in kwargs: update_fields.append('error_message = ?') values.append(kwargs['error_message']) if not update_fields: return False values.append(build_id) query = f"UPDATE builds SET {', '.join(update_fields)} WHERE build_id = ?" cursor.execute(query, values) conn.commit() conn.close() return True except Exception as e: print(f"Failed to update build status: {e}") return False def get_build(self, build_id: str) -> Optional[BuildRecord]: """Get a specific build record""" try: with self.lock: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute('SELECT * FROM builds WHERE build_id = ?', (build_id,)) row = cursor.fetchone() conn.close() if row: return self._row_to_build_record(row) return None except Exception as e: print(f"Failed to get build record: {e}") return None def get_builds_by_blueprint(self, blueprint: str, limit: Optional[int] = None) -> List[BuildRecord]: """Get builds by blueprint name""" try: with self.lock: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() query = 'SELECT * FROM builds WHERE blueprint = ? ORDER BY created_at DESC' if limit: query += f' LIMIT {limit}' cursor.execute(query, (blueprint,)) rows = cursor.fetchall() conn.close() return [self._row_to_build_record(row) for row in rows] except Exception as e: print(f"Failed to get builds by blueprint: {e}") return [] def get_builds_by_status(self, status: str, limit: Optional[int] = None) -> List[BuildRecord]: """Get builds by status""" try: with self.lock: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() query = 'SELECT * FROM builds WHERE status = ? ORDER BY created_at DESC' if limit: query += f' LIMIT {limit}' cursor.execute(query, (status,)) rows = cursor.fetchall() conn.close() return [self._row_to_build_record(row) for row in rows] except Exception as e: print(f"Failed to get builds by status: {e}") return [] def get_recent_builds(self, limit: int = 50) -> List[BuildRecord]: """Get recent builds""" try: with self.lock: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute('SELECT * FROM builds ORDER BY created_at DESC LIMIT ?', (limit,)) rows = cursor.fetchall() conn.close() return [self._row_to_build_record(row) for row in rows] except Exception as e: print(f"Failed to get recent builds: {e}") return [] def get_build_statistics(self) -> Dict[str, Any]: """Get build statistics""" try: with self.lock: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() # Total builds cursor.execute('SELECT COUNT(*) FROM builds') total_builds = cursor.fetchone()[0] # Builds by status cursor.execute('SELECT status, COUNT(*) FROM builds GROUP BY status') status_counts = dict(cursor.fetchall()) # Builds by blueprint cursor.execute('SELECT blueprint, COUNT(*) FROM builds GROUP BY blueprint') blueprint_counts = dict(cursor.fetchall()) # Average duration cursor.execute('SELECT AVG(duration) FROM builds WHERE duration IS NOT NULL') avg_duration = cursor.fetchone()[0] or 0 # Success rate cursor.execute('SELECT COUNT(*) FROM builds WHERE status = "FINISHED"') successful_builds = cursor.fetchone()[0] success_rate = (successful_builds / total_builds * 100) if total_builds > 0 else 0 conn.close() return { 'total_builds': total_builds, 'status_counts': status_counts, 'blueprint_counts': blueprint_counts, 'average_duration': avg_duration, 'success_rate': success_rate, 'successful_builds': successful_builds } except Exception as e: print(f"Failed to get build statistics: {e}") return {} def _row_to_build_record(self, row) -> BuildRecord: """Convert database row to BuildRecord""" return BuildRecord( build_id=row[0], blueprint=row[1], target=row[2], architecture=row[3], status=row[4], created_at=datetime.fromisoformat(row[5]), completed_at=datetime.fromisoformat(row[6]) if row[6] else None, duration=row[7], metadata=json.loads(row[8]) if row[8] else {}, logs=json.loads(row[9]) if row[9] else [], artifacts=json.loads(row[10]) if row[10] else [], error_message=row[11] ) class BuildHistoryManager: """High-level build history management""" def __init__(self, db_path: str = "build_history.db"): self.db = BuildHistoryDB(db_path) self.active_builds: Dict[str, BuildRecord] = {} def start_build(self, build_id: str, blueprint: str, target: str, architecture: str, metadata: Optional[Dict] = None) -> bool: """Start tracking a new build""" build_record = BuildRecord( build_id=build_id, blueprint=blueprint, target=target, architecture=architecture, status="RUNNING", created_at=datetime.now(), completed_at=None, duration=None, metadata=metadata or {}, logs=[], artifacts=[], error_message=None ) # Add to database if self.db.add_build(build_record): self.active_builds[build_id] = build_record return True return False def update_build_progress(self, build_id: str, status: str, logs: Optional[List[str]] = None, artifacts: Optional[List[str]] = None) -> bool: """Update build progress""" if build_id in self.active_builds: build_record = self.active_builds[build_id] # Update fields update_data = {'status': status} if logs is not None: build_record.logs.extend(logs) update_data['logs'] = build_record.logs if artifacts is not None: build_record.artifacts.extend(artifacts) update_data['artifacts'] = build_record.artifacts # Update completion time and duration if finished if status in ["FINISHED", "FAILED"]: build_record.completed_at = datetime.now() build_record.duration = (build_record.completed_at - build_record.created_at).total_seconds() update_data['completed_at'] = build_record.completed_at update_data['duration'] = build_record.duration # Remove from active builds del self.active_builds[build_id] # Update database return self.db.update_build_status(build_id, **update_data) return False def get_build_summary(self) -> Dict[str, Any]: """Get build summary information""" stats = self.db.get_build_statistics() stats['active_builds'] = len(self.active_builds) stats['active_build_ids'] = list(self.active_builds.keys()) return stats def export_history(self, output_path: str, format: str = "json") -> bool: """Export build history to file""" try: builds = self.db.get_recent_builds(limit=1000) # Export all builds if format.lower() == "json": with open(output_path, 'w') as f: json.dump([asdict(build) for build in builds], f, indent=2, default=str) else: print(f"Unsupported export format: {format}") return False return True except Exception as e: print(f"Failed to export history: {e}") return False def main(): """Example usage of build history""" print("Build History Example") print("This module provides build history tracking for composer builds") if __name__ == '__main__': main()