"""
try:
# Get the aggregation data - now it's directly the current_results
aggregation_data = self.current_results
# Handle different result types
if isinstance(aggregation_data, str):
try:
# Try to parse as JSON first
parsed_data = json.loads(aggregation_data)
data_to_encode = parsed_data
except json.JSONDecodeError:
# If it's not valid JSON, wrap it in a dict
data_to_encode = {"aggregation_output": aggregation_data}
elif hasattr(aggregation_data, 'to_dict'):
# Handle AgentResult objects
data_to_encode = aggregation_data.to_dict()
elif isinstance(aggregation_data, dict):
data_to_encode = aggregation_data
else:
# Fallback for other object types
data_to_encode = {"aggregation_output": str(aggregation_data)}
# Format JSON for display
formatted_json = json.dumps(data_to_encode, indent=2)
return f"""
✅ Analysis Complete! Ready for Visualization
📋 Steps to visualize your results:
1. Click "Open JSONCrack Editor" below
2. Click "Copy JSON" button or click the JSON data below to select all
3. Paste it into the JSONCrack editor
"
def get_logs_html(self) -> str:
"""Generate HTML for live logs display"""
if self.current_agent_name is None:
return "
No agent initialized yet
"
return f"""
📝 Logging Status for Agent: {self.current_agent_name}
✅ Standard Python Logging Active
• All logs are being captured by the application's logging system
• Check your console/terminal for real-time log output
• Logs include detailed information about agent execution
• Structured logging with timestamps and log levels
📋 Log Types Available:
• INFO - General information and progress
• DEBUG - Detailed debugging information
• WARNING - Warning messages
• ERROR - Error messages
🔍 What You'll See:
• Agent initialization and configuration
• MCP tool interactions and responses
• Analysis progress and completion status
• Any errors or warnings during execution
"""
def test_log_writing(self):
"""Test function to write a sample log entry"""
if self.current_agent_name:
try:
write_lineage_log(self.current_agent_name, "test", "Test log entry from frontend")
self.logger.info(f"Test log written successfully for agent: {self.current_agent_name}")
return f"✅ Test log written successfully for agent: {self.current_agent_name}! Check your console output."
except Exception as e:
self.logger.error(f"Failed to write test log: {e}")
return f"❌ Failed to write test log: {e}"
else:
return "⚠️ Please initialize an agent first by running an analysis"
def get_results_info(self) -> str:
"""Get information about the current results"""
if self.current_results is None:
return "No results available yet"
if isinstance(self.current_results, dict) and "error" in self.current_results:
return f"Error in results: {self.current_results['error']}"
if hasattr(self.current_results, 'to_dict'):
# AgentResult object
result_dict = self.current_results.to_dict()
inputs_count = len(result_dict.get('inputs', []))
outputs_count = len(result_dict.get('outputs', []))
return f"✅ Structured results with {inputs_count} input(s) and {outputs_count} output(s)"
if isinstance(self.current_results, dict):
return f"✅ Dictionary results with {len(self.current_results)} keys"
return f"✅ Results type: {type(self.current_results).__name__}"
async def run_analysis(self, agent_name: str, model_name: str, query: str):
"""Run SQL lineage analysis"""
try:
# Validate input
if not query or not query.strip():
return "❌ Error: Query cannot be empty. Please provide a valid query for analysis."
self.logger.info(f"Starting analysis with agent: {agent_name}, model: {model_name}")
# Initialize the agent framework with simplified constructor
self.agent_framework = FrameworkAgent(
agent_name=agent_name,
model_name=model_name,
source_code=query.strip()
)
self.current_agent_name = agent_name
self.logger.info(f"Agent framework initialized. Running analysis...")
# Run the analysis using the structured results method
results = await self.agent_framework.run_agent()
self.current_results = results
# Check if we got an error response
if isinstance(results, dict) and "error" in results:
self.logger.error(f"Analysis failed: {results['error']}")
return f"❌ Analysis failed: {results['error']}"
self.logger.info(f"Analysis completed successfully for agent: {agent_name}")
return f"""✅ Analysis completed successfully! Results are now available in the visualization section.
Click 'Open JSONCrack Editor' to visualize your data lineage.
If you want to set up your own local development environment or deploy this in production,
please refer to the GitHub repository mentioned above."""
except ValueError as ve:
self.logger.error(f"Validation error: {ve}")
return f"❌ Validation error: {str(ve)}"
except Exception as e:
self.logger.error(f"Error running analysis: {e}")
return f"❌ Error running analysis: {str(e)}"
def run_analysis_sync(self, agent_name: str, model_name: str, query: str):
"""Synchronous wrapper for run_analysis"""
return asyncio.run(self.run_analysis(agent_name, model_name, query))
def create_ui(self):
"""Create the Gradio interface"""
with gr.Blocks(title="SQL Lineage Analysis", fill_width=True) as ui:
gr.Markdown('