""" Report Integration for AegisLM Framework Production-grade integration between reporting, experiment, and audit systems. """ import uuid from datetime import datetime from typing import Dict, Any, Optional, List from pathlib import Path import sys # Add parent directory to path for imports current_dir = Path(__file__).parent backend_dir = current_dir.parent if str(backend_dir) not in sys.path: sys.path.insert(0, str(backend_dir)) from reporting.report_generator import get_report_generator from reporting.export_utils import get_report_exporter from reporting.report_validator import get_report_validator from reporting.file_manager import get_file_manager from schemas.report_schema import ( ReportGenerationRequest, ReportGenerationResult, ReportFormat, ReportType ) from experiments.experiment_manager import get_experiment_manager from audit.audit_manager import get_audit_manager class ReportIntegrationService: """ Production-grade integration service for reporting system. Connects reporting with experiment and audit systems. """ def __init__(self): """Initialize integration service.""" self.report_generator = get_report_generator() self.report_exporter = get_report_exporter() self.report_validator = get_report_validator() self.file_manager = get_file_manager() self.experiment_manager = get_experiment_manager() self.audit_manager = get_audit_manager() def generate_comprehensive_report(self, run_id: str, report_type: ReportType = ReportType.FULL, format: ReportFormat = ReportFormat.JSON, requested_by: Optional[str] = None) -> ReportGenerationResult: """ Generate comprehensive report with full integration. Args: run_id: Run ID report_type: Type of report format: Export format requested_by: User requesting report Returns: Report generation result """ try: # Verify experiment exists experiment = self.experiment_manager.get_experiment(run_id) if not experiment: return ReportGenerationResult( request=ReportGenerationRequest( run_id=run_id, report_type=report_type, format=format, requested_by=requested_by ), success=False, generation_started_at=datetime.utcnow(), generation_completed_at=datetime.utcnow(), generation_duration_ms=0, status="failed", error_message=f"Experiment with run_id {run_id} not found" ) # Verify audit trail exists audit_trail = self.audit_manager.get_audit_trail(run_id) if not audit_trail: return ReportGenerationResult( request=ReportGenerationRequest( run_id=run_id, report_type=report_type, format=format, requested_by=requested_by ), success=False, generation_started_at=datetime.utcnow(), generation_completed_at=datetime.utcnow(), generation_duration_ms=0, status="failed", error_message=f"Audit trail for run_id {run_id} not found" ) # Generate report request request = ReportGenerationRequest( run_id=run_id, report_type=report_type, format=format, include_full_config=True, include_full_result=True, include_analysis=True, requested_by=requested_by ) # Generate report result = self.report_generator.generate_report(request) if result.success and result.report_id: # Export report to file cached_report = self.report_generator.get_report(result.report_id) if cached_report: if format == ReportFormat.JSON: file_path = self.report_exporter.export_json_report( cached_report['data'], run_id, report_type ) elif format == ReportFormat.CSV: if report_type == ReportType.SUMMARY: file_path = self.report_exporter.export_csv_summary( cached_report['data'], run_id ) else: file_path = self.report_exporter.export_csv_detailed( cached_report['data'], run_id ) # Update result with file information result.file_path = file_path result.file_size_bytes = Path(file_path).stat().st_size result.file_checksum = self.report_validator._calculate_content_checksum( Path(file_path).read_text() ) result.integrity_verified = True return result except Exception as e: return ReportGenerationResult( request=ReportGenerationRequest( run_id=run_id, report_type=report_type, format=format, requested_by=requested_by ), success=False, generation_started_at=datetime.utcnow(), generation_completed_at=datetime.utcnow(), generation_duration_ms=0, status="failed", error_message=f"Integration error: {str(e)}" ) def generate_batch_reports(self, run_ids: List[str], report_type: ReportType = ReportType.SUMMARY, format: ReportFormat = ReportFormat.JSON, requested_by: Optional[str] = None) -> List[ReportGenerationResult]: """ Generate reports for multiple runs with validation. Args: run_ids: List of run IDs report_type: Type of report format: Export format requested_by: User requesting reports Returns: List of report generation results """ results = [] for run_id in run_ids: try: result = self.generate_comprehensive_report(run_id, report_type, format, requested_by) results.append(result) except Exception as e: # Create failed result for this run results.append(ReportGenerationResult( request=ReportGenerationRequest( run_id=run_id, report_type=report_type, format=format, requested_by=requested_by ), success=False, generation_started_at=datetime.utcnow(), generation_completed_at=datetime.utcnow(), generation_duration_ms=0, status="failed", error_message=f"Batch generation error: {str(e)}" )) return results def verify_report_integrity(self, file_path: str) -> Dict[str, Any]: """ Verify report integrity with full system integration. Args: file_path: Path to report file Returns: Integrity verification result """ try: # Basic file validation validation_request = ReportValidationRequest(file_path=file_path) validation_result = self.report_validator.verify_report(validation_request) # Extract run_id from report run_id = None if validation_result.report_data: experiment_data = validation_result.report_data.get('experiment', {}) run_id = experiment_data.get('run_id') # Cross-validate with experiment system experiment_valid = True audit_valid = True if run_id: # Verify experiment exists experiment = self.experiment_manager.get_experiment(run_id) experiment_valid = experiment is not None # Verify audit trail exists audit_trail = self.audit_manager.get_audit_trail(run_id) audit_valid = audit_trail is not None # Calculate overall integrity score integrity_score = 0.0 if validation_result.is_valid: integrity_score += 0.4 if validation_result.integrity_valid: integrity_score += 0.3 if experiment_valid: integrity_score += 0.15 if audit_valid: integrity_score += 0.15 return { "file_path": file_path, "validation_result": validation_result, "cross_validation": { "experiment_valid": experiment_valid, "audit_valid": audit_valid, "run_id": run_id }, "integrity_score": round(integrity_score, 3), "overall_valid": validation_result.is_valid and experiment_valid and audit_valid, "verification_timestamp": datetime.utcnow().isoformat() } except Exception as e: return { "file_path": file_path, "error": f"Integrity verification failed: {str(e)}", "integrity_score": 0.0, "overall_valid": False, "verification_timestamp": datetime.utcnow().isoformat() } def get_run_report_summary(self, run_id: str) -> Dict[str, Any]: """ Get comprehensive summary of reports for a run. Args: run_id: Run ID Returns: Report summary """ try: # Get experiment info experiment = self.experiment_manager.get_experiment(run_id) experiment_info = None if experiment: experiment_info = { "run_id": str(experiment.run_id), "experiment_name": experiment.experiment_name, "status": experiment.status.value, "created_at": experiment.created_at.isoformat(), "completed_at": experiment.completed_at.isoformat() if experiment.completed_at else None, "model_name": experiment.config_snapshot.model_name, "dataset_name": experiment.config_snapshot.dataset_name, "attack_types": experiment.config_snapshot.attack_types } # Get audit info audit_trail = self.audit_manager.get_audit_trail(run_id) audit_info = None if audit_trail: audit_info = { "run_id": audit_trail.run_id, "status": audit_trail.status.value, "integrity_level": audit_trail.integrity_level.value, "config_hash": audit_trail.config_record.config_hash, "result_checksum": audit_trail.result_record.result_checksum, "replay_count": audit_trail.replay_count, "verified_at": audit_trail.verified_at.isoformat() if audit_trail.verified_at else None } # Get available reports reports = self.report_exporter.list_reports(run_id=run_id) # Group reports by type report_summary = { "total_reports": len(reports), "json_reports": len([r for r in reports if r.get('file_extension') == '.json']), "csv_reports": len([r for r in reports if r.get('file_extension') == '.csv']), "reports_by_type": {} } for report in reports: filename = report.get('file_name', '') if 'full' in filename.lower(): report_summary["reports_by_type"]["full"] = report_summary["reports_by_type"].get("full", 0) + 1 elif 'summary' in filename.lower(): report_summary["reports_by_type"]["summary"] = report_summary["reports_by_type"].get("summary", 0) + 1 elif 'executive' in filename.lower(): report_summary["reports_by_type"]["executive"] = report_summary["reports_by_type"].get("executive", 0) + 1 elif 'technical' in filename.lower(): report_summary["reports_by_type"]["technical"] = report_summary["reports_by_type"].get("technical", 0) + 1 elif 'compliance' in filename.lower(): report_summary["reports_by_type"]["compliance"] = report_summary["reports_by_type"].get("compliance", 0) + 1 # Get latest report info latest_report = reports[0] if reports else None return { "run_id": run_id, "experiment_info": experiment_info, "audit_info": audit_info, "report_summary": report_summary, "latest_report": latest_report, "available_reports": reports, "summary_timestamp": datetime.utcnow().isoformat() } except Exception as e: return { "run_id": run_id, "error": f"Failed to get report summary: {str(e)}", "summary_timestamp": datetime.utcnow().isoformat() } def export_run_package(self, run_id: str, include_all_formats: bool = True, requested_by: Optional[str] = None) -> Dict[str, Any]: """ Export complete package for a run including all report formats. Args: run_id: Run ID include_all_formats: Whether to include all formats requested_by: User requesting export Returns: Export package information """ try: # Verify run exists experiment = self.experiment_manager.get_experiment(run_id) if not experiment: return {"error": f"Experiment with run_id {run_id} not found"} audit_trail = self.audit_manager.get_audit_trail(run_id) if not audit_trail: return {"error": f"Audit trail for run_id {run_id} not found"} # Generate reports in different formats report_types = [ReportType.FULL, ReportType.SUMMARY, ReportType.EXECUTIVE] formats = [ReportFormat.JSON, ReportFormat.CSV] if include_all_formats else [ReportFormat.JSON] exported_files = [] export_results = [] for report_type in report_types: for format in formats: try: result = self.generate_comprehensive_report( run_id, report_type, format, requested_by ) if result.success and result.file_path: exported_files.append({ "file_path": result.file_path, "file_name": Path(result.file_path).name, "report_type": report_type.value, "format": format.value, "file_size_bytes": result.file_size_bytes, "file_checksum": result.file_checksum, "integrity_verified": result.integrity_verified }) export_results.append({ "report_type": report_type.value, "format": format.value, "success": result.success, "error": result.error_message if not result.success else None }) except Exception as e: export_results.append({ "report_type": report_type.value, "format": format.value, "success": False, "error": str(e) }) # Create package metadata package_metadata = { "run_id": run_id, "export_timestamp": datetime.utcnow().isoformat(), "requested_by": requested_by, "experiment_info": { "run_id": str(experiment.run_id), "experiment_name": experiment.experiment_name, "model_name": experiment.config_snapshot.model_name, "dataset_name": experiment.config_snapshot.dataset_name, "status": experiment.status.value }, "audit_info": { "integrity_level": audit_trail.integrity_level.value, "config_hash": audit_trail.config_record.config_hash, "result_checksum": audit_trail.result_record.result_checksum }, "export_summary": { "total_files": len(exported_files), "successful_exports": len([f for f in exported_files if f.get("integrity_verified")]), "total_size_mb": round(sum(f.get("file_size_bytes", 0) for f in exported_files) / (1024 * 1024), 2), "formats_included": list(set(f.get("format") for f in exported_files)), "report_types_included": list(set(f.get("report_type") for f in exported_files)) }, "exported_files": exported_files, "export_results": export_results } return package_metadata except Exception as e: return {"error": f"Failed to export run package: {str(e)}"} def validate_run_completeness(self, run_id: str) -> Dict[str, Any]: """ Validate that a run has complete reporting coverage. Args: run_id: Run ID Returns: Completeness validation result """ try: # Check experiment completeness experiment = self.experiment_manager.get_experiment(run_id) experiment_complete = experiment is not None and experiment.status.value == "completed" # Check audit completeness audit_trail = self.audit_manager.get_audit_trail(run_id) audit_complete = audit_trail is not None and audit_trail.status.value == "verified" # Check report coverage reports = self.report_exporter.list_reports(run_id=run_id) report_types_present = set() formats_present = set() for report in reports: filename = report.get('file_name', '').lower() if 'full' in filename: report_types_present.add('full') if 'summary' in filename: report_types_present.add('summary') if 'executive' in filename: report_types_present.add('executive') ext = report.get('file_extension', '').replace('.', '') if ext: formats_present.add(ext) # Required report types required_types = {'full', 'summary'} required_formats = {'json'} reports_complete = ( required_types.issubset(report_types_present) and required_formats.issubset(formats_present) ) # Calculate completeness score completeness_score = 0.0 if experiment_complete: completeness_score += 0.4 if audit_complete: completeness_score += 0.3 if reports_complete: completeness_score += 0.3 # Identify missing items missing_items = [] if not experiment_complete: missing_items.append("completed experiment") if not audit_complete: missing_items.append("verified audit trail") if not required_types.issubset(report_types_present): missing_types = required_types - report_types_present missing_items.extend([f"{t} report" for t in missing_types]) if not required_formats.issubset(formats_present): missing_formats = required_formats - formats_present missing_items.extend([f"{f} format" for f in missing_formats]) return { "run_id": run_id, "completeness_score": round(completeness_score, 3), "overall_complete": completeness_score >= 0.9, "validation_timestamp": datetime.utcnow().isoformat(), "components": { "experiment_complete": experiment_complete, "audit_complete": audit_complete, "reports_complete": reports_complete }, "report_coverage": { "report_types_present": list(report_types_present), "formats_present": list(formats_present), "required_types": list(required_types), "required_formats": list(required_formats) }, "missing_items": missing_items, "recommendations": self._generate_completeness_recommendations(completeness_score, missing_items) } except Exception as e: return { "run_id": run_id, "error": f"Completeness validation failed: {str(e)}", "completeness_score": 0.0, "overall_complete": False, "validation_timestamp": datetime.utcnow().isoformat() } def _generate_completeness_recommendations(self, score: float, missing_items: List[str]) -> List[str]: """Generate recommendations based on completeness score.""" recommendations = [] if score < 0.5: recommendations.append("Run is significantly incomplete - address all missing components") elif score < 0.8: recommendations.append("Run has some gaps - complete missing items for full coverage") if "completed experiment" in missing_items: recommendations.append("Wait for experiment to complete or check experiment status") if "verified audit trail" in missing_items: recommendations.append("Run audit verification to ensure integrity") for item in missing_items: if "report" in item: recommendations.append(f"Generate {item} for complete documentation") elif "format" in item: recommendations.append(f"Export report in {item} for broader accessibility") if not recommendations: recommendations.append("Run is complete - all reporting requirements met") return recommendations # Global instance for easy access _integration_service = None def get_report_integration_service() -> ReportIntegrationService: """ Get the global report integration service instance. Returns: Global report integration service instance """ global _integration_service if _integration_service is None: _integration_service = ReportIntegrationService() return _integration_service # Convenience functions for direct usage def generate_run_report(run_id: str, report_type: ReportType = ReportType.FULL, format: ReportFormat = ReportFormat.JSON, requested_by: Optional[str] = None) -> ReportGenerationResult: """ Generate comprehensive report for a run. Args: run_id: Run ID report_type: Type of report format: Export format requested_by: User requesting report Returns: Report generation result """ service = get_report_integration_service() return service.generate_comprehensive_report(run_id, report_type, format, requested_by) def verify_report_package(file_path: str) -> Dict[str, Any]: """ Verify report package integrity. Args: file_path: Path to report file Returns: Integrity verification result """ service = get_report_integration_service() return service.verify_report_integrity(file_path)