chanages

src/streamlit_app.py

@@ -24,7 +24,9 @@ try:
         classify_data_structure,
         get_hierarchy_summary
     )
-except ImportError:
+    st.success("✅ Successfully imported structure_analysis module")
+except ImportError as e:
+    st.error(f"⚠️ Import error: {e}")
     st.error("⚠️ structure_analysis.py not found. Make sure all files are uploaded.")
     st.stop()
 
@@ -38,25 +40,45 @@ def analyze_with_llm(data: Dict[str, Any], target_field: str = "rotation_enabled
     Analyze data and generate a prompt for LLM analysis.
     Returns structured analysis without requiring Ollama.
     """
+    print(f"DEBUG: Starting analysis with target_field: {target_field}")
+    print(f"DEBUG: Data type: {type(data)}")
+    print(f"DEBUG: Data keys: {list(data.keys()) if isinstance(data, dict) else 'Not a dict'}")
+    
     # Detect summary fields
+    print("DEBUG: Detecting summary fields...")
     summary_fields = detect_summary_fields(data)
+    print(f"DEBUG: Found summary fields: {summary_fields}")
+    
+    print("DEBUG: Classifying data structure...")
     classification = classify_data_structure(data)
+    print(f"DEBUG: Classification result: {classification}")
+    
+    print("DEBUG: Getting hierarchy summary...")
     hierarchy_summary = get_hierarchy_summary(data)
+    print(f"DEBUG: Hierarchy summary: {hierarchy_summary}")
     
     # Extract samples
+    print("DEBUG: Extracting samples...")
     sample_object = {}
     if 'results' in data:
-        for section in data['results'].values():
+        print("DEBUG: Found 'results' key in data")
+        for section_name, section in data['results'].items():
+            print(f"DEBUG: Processing section '{section_name}': {type(section)}")
             if isinstance(section, list) and len(section) > 0:
                 sample_object = section[0]
+                print(f"DEBUG: Found sample object from list: {sample_object}")
                 break
             elif isinstance(section, dict):
                 for key, value in section.items():
                     if isinstance(value, list) and len(value) > 0:
                         sample_object = value[0] if isinstance(value[0], dict) else {}
+                        print(f"DEBUG: Found sample object from dict list: {sample_object}")
                         break
+    else:
+        print("DEBUG: No 'results' key found in data")
     
     summary_sample = data.get('results', {}).get('summary', {}) or data.get('summary', {})
+    print(f"DEBUG: Summary sample: {summary_sample}")
     
     # Count objects with target field
     def count_objects_with_field(obj, field_name):
@@ -71,9 +93,12 @@ def analyze_with_llm(data: Dict[str, Any], target_field: str = "rotation_enabled
                 count += count_objects_with_field(item, field_name)
         return count
     
+    print("DEBUG: Counting objects with target field...")
     total_objects = count_objects_with_field(data, target_field)
+    print(f"DEBUG: Total objects with '{target_field}': {total_objects}")
     
     # Generate analysis
+    print("DEBUG: Generating analysis...")
     analysis = {
         "summary_fields_detected": summary_fields[:10],
         "classification": classification,
@@ -83,15 +108,23 @@ def analyze_with_llm(data: Dict[str, Any], target_field: str = "rotation_enabled
         "summary_sample": summary_sample,
         "recommended_fields": []
     }
+    print(f"DEBUG: Initial analysis: {analysis}")
     
     # Recommend fields based on priority
+    print("DEBUG: Generating field recommendations...")
     if summary_fields:
         analysis["recommended_fields"].extend(summary_fields[:3])
+        print(f"DEBUG: Added summary fields: {summary_fields[:3]}")
     if classification.get('config_fields'):
         analysis["recommended_fields"].extend(classification['config_fields'][:2])
+        print(f"DEBUG: Added config fields: {classification['config_fields'][:2]}")
     if sample_object:
-        analysis["recommended_fields"].extend([k for k in sample_object.keys() if target_field in k.lower()])
+        target_related = [k for k in sample_object.keys() if target_field in k.lower()]
+        analysis["recommended_fields"].extend(target_related)
+        print(f"DEBUG: Added target-related fields: {target_related}")
     
+    print(f"DEBUG: Final recommended fields: {analysis['recommended_fields']}")
+    print("DEBUG: Analysis completed successfully")
     return analysis
 
 
@@ -146,12 +179,20 @@ def main():
     )
     
     if uploaded_file is not None:
+        # Debug file upload info
+        st.info(f"📁 File uploaded: {uploaded_file.name} (Size: {uploaded_file.size} bytes)")
+        
         # Read and parse JSON
         try:
+            st.info("🔄 Reading file content...")
             content = uploaded_file.read()
+            st.info(f"📄 Content length: {len(content)} characters")
+            
+            st.info("🔄 Parsing JSON...")
             data = json.loads(content)
             
             st.success("✅ File loaded successfully!")
+            st.info(f"📊 Data structure: {type(data)} with {len(data) if isinstance(data, (dict, list)) else 'unknown'} top-level items")
             
             # Sidebar for settings
             with st.sidebar:
@@ -167,9 +208,17 @@ def main():
                 # Analyze button
                 if st.button("🔍 Analyze", type="primary"):
                     with st.spinner("Analyzing data structure..."):
-                        analysis_result = analyze_with_llm(data, target_field)
-                        st.session_state.analysis_result = analysis_result
-                        st.session_state.data = data
+                        st.info(f"🎯 Analyzing with target field: {target_field}")
+                        try:
+                            analysis_result = analyze_with_llm(data, target_field)
+                            st.session_state.analysis_result = analysis_result
+                            st.session_state.data = data
+                            st.success("✅ Analysis completed successfully!")
+                        except Exception as e:
+                            st.error(f"❌ Analysis failed: {e}")
+                            st.error(f"Error type: {type(e).__name__}")
+                            import traceback
+                            st.code(traceback.format_exc())
             
             # Display results if available
             if st.session_state.analysis_result:

test_analysis.py

@@ -0,0 +1,151 @@
+#!/usr/bin/env python3
+"""
+Test script to verify the analysis functions work correctly without Streamlit.
+"""
+
+import json
+import sys
+import os
+
+# Add src directory to path
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src'))
+
+from structure_analysis import detect_summary_fields, classify_data_structure, get_hierarchy_summary
+
+def analyze_with_llm(data, target_field="rotation_enabled"):
+    """
+    Analyze data and generate a prompt for LLM analysis.
+    Returns structured analysis without requiring Ollama.
+    """
+    print(f"DEBUG: Starting analysis with target_field: {target_field}")
+    print(f"DEBUG: Data type: {type(data)}")
+    print(f"DEBUG: Data keys: {list(data.keys()) if isinstance(data, dict) else 'Not a dict'}")
+    
+    # Detect summary fields
+    print("DEBUG: Detecting summary fields...")
+    summary_fields = detect_summary_fields(data)
+    print(f"DEBUG: Found summary fields: {summary_fields}")
+    
+    print("DEBUG: Classifying data structure...")
+    classification = classify_data_structure(data)
+    print(f"DEBUG: Classification result: {classification}")
+    
+    print("DEBUG: Getting hierarchy summary...")
+    hierarchy_summary = get_hierarchy_summary(data)
+    print(f"DEBUG: Hierarchy summary: {hierarchy_summary}")
+    
+    # Extract samples
+    print("DEBUG: Extracting samples...")
+    sample_object = {}
+    if 'results' in data:
+        print("DEBUG: Found 'results' key in data")
+        for section_name, section in data['results'].items():
+            print(f"DEBUG: Processing section '{section_name}': {type(section)}")
+            if isinstance(section, list) and len(section) > 0:
+                sample_object = section[0]
+                print(f"DEBUG: Found sample object from list: {sample_object}")
+                break
+            elif isinstance(section, dict):
+                for key, value in section.items():
+                    if isinstance(value, list) and len(value) > 0:
+                        sample_object = value[0] if isinstance(value[0], dict) else {}
+                        print(f"DEBUG: Found sample object from dict list: {sample_object}")
+                        break
+    else:
+        print("DEBUG: No 'results' key found in data")
+    
+    summary_sample = data.get('results', {}).get('summary', {}) or data.get('summary', {})
+    print(f"DEBUG: Summary sample: {summary_sample}")
+    
+    # Count objects with target field
+    def count_objects_with_field(obj, field_name):
+        count = 0
+        if isinstance(obj, dict):
+            if field_name in obj:
+                count += 1
+            for v in obj.values():
+                count += count_objects_with_field(v, field_name)
+        elif isinstance(obj, list):
+            for item in obj:
+                count += count_objects_with_field(item, field_name)
+        return count
+    
+    print("DEBUG: Counting objects with target field...")
+    total_objects = count_objects_with_field(data, target_field)
+    print(f"DEBUG: Total objects with '{target_field}': {total_objects}")
+    
+    # Generate analysis
+    print("DEBUG: Generating analysis...")
+    analysis = {
+        "summary_fields_detected": summary_fields[:10],
+        "classification": classification,
+        "hierarchy_summary": hierarchy_summary,
+        "total_objects": total_objects,
+        "sample_object": sample_object,
+        "summary_sample": summary_sample,
+        "recommended_fields": []
+    }
+    print(f"DEBUG: Initial analysis: {analysis}")
+    
+    # Recommend fields based on priority
+    print("DEBUG: Generating field recommendations...")
+    if summary_fields:
+        analysis["recommended_fields"].extend(summary_fields[:3])
+        print(f"DEBUG: Added summary fields: {summary_fields[:3]}")
+    if classification.get('config_fields'):
+        analysis["recommended_fields"].extend(classification['config_fields'][:2])
+        print(f"DEBUG: Added config fields: {classification['config_fields'][:2]}")
+    if sample_object:
+        target_related = [k for k in sample_object.keys() if target_field in k.lower()]
+        analysis["recommended_fields"].extend(target_related)
+        print(f"DEBUG: Added target-related fields: {target_related}")
+    
+    print(f"DEBUG: Final recommended fields: {analysis['recommended_fields']}")
+    print("DEBUG: Analysis completed successfully")
+    return analysis
+
+def test_analysis_functions():
+    """Test the analysis functions with sample data."""
+    
+    # Load test data
+    with open('test_data.json', 'r') as f:
+        data = json.load(f)
+    
+    print("=== Testing Analysis Functions ===")
+    print(f"Data loaded: {type(data)}")
+    print(f"Data keys: {list(data.keys())}")
+    print()
+    
+    # Test detect_summary_fields
+    print("Testing detect_summary_fields...")
+    summary_fields = detect_summary_fields(data)
+    print(f"Summary fields found: {summary_fields}")
+    print()
+    
+    # Test classify_data_structure
+    print("Testing classify_data_structure...")
+    classification = classify_data_structure(data)
+    print(f"Classification result: {classification}")
+    print()
+    
+    # Test get_hierarchy_summary
+    print("Testing get_hierarchy_summary...")
+    hierarchy_summary = get_hierarchy_summary(data)
+    print(f"Hierarchy summary: {hierarchy_summary}")
+    print()
+    
+    # Test the full analysis function
+    print("Testing full analysis...")
+    try:
+        analysis_result = analyze_with_llm(data, "rotation_enabled")
+        print("✅ Analysis completed successfully!")
+        print(f"Analysis keys: {list(analysis_result.keys())}")
+        print(f"Recommended fields: {analysis_result.get('recommended_fields', [])}")
+        print(f"Total objects with 'rotation_enabled': {analysis_result.get('total_objects', 0)}")
+    except Exception as e:
+        print(f"❌ Analysis failed: {e}")
+        import traceback
+        traceback.print_exc()
+
+if __name__ == "__main__":
+    test_analysis_functions()

test_data.json

@@ -0,0 +1,31 @@
+{
+  "results": {
+    "summary": {
+      "total_keys": 13,
+      "rotated_keys": 6,
+      "rotation_percentage": 46,
+      "compliance_rate": 85
+    },
+    "kms_keys": {
+      "object": [
+        {
+          "key_id": "12345",
+          "rotation_enabled": true,
+          "key_state": "Enabled",
+          "creation_date": "2023-01-15"
+        },
+        {
+          "key_id": "67890",
+          "rotation_enabled": false,
+          "key_state": "Enabled",
+          "creation_date": "2023-02-20"
+        }
+      ]
+    },
+    "config": {
+      "rotation_policy": "enabled",
+      "compliance_enforced": true,
+      "audit_enabled": true
+    }
+  }
+}

test_upload.py

@@ -0,0 +1,59 @@
+#!/usr/bin/env python3
+"""
+Test script to verify the file upload functionality works correctly.
+"""
+
+import json
+import sys
+import os
+
+# Add src directory to path
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src'))
+
+from structure_analysis import detect_summary_fields, classify_data_structure, get_hierarchy_summary
+
+def test_analysis_functions():
+    """Test the analysis functions with sample data."""
+    
+    # Load test data
+    with open('test_data.json', 'r') as f:
+        data = json.load(f)
+    
+    print("=== Testing Analysis Functions ===")
+    print(f"Data loaded: {type(data)}")
+    print(f"Data keys: {list(data.keys())}")
+    print()
+    
+    # Test detect_summary_fields
+    print("Testing detect_summary_fields...")
+    summary_fields = detect_summary_fields(data)
+    print(f"Summary fields found: {summary_fields}")
+    print()
+    
+    # Test classify_data_structure
+    print("Testing classify_data_structure...")
+    classification = classify_data_structure(data)
+    print(f"Classification result: {classification}")
+    print()
+    
+    # Test get_hierarchy_summary
+    print("Testing get_hierarchy_summary...")
+    hierarchy_summary = get_hierarchy_summary(data)
+    print(f"Hierarchy summary: {hierarchy_summary}")
+    print()
+    
+    # Test the full analysis function
+    print("Testing full analysis...")
+    try:
+        from streamlit_app import analyze_with_llm
+        analysis_result = analyze_with_llm(data, "rotation_enabled")
+        print("✅ Analysis completed successfully!")
+        print(f"Analysis keys: {list(analysis_result.keys())}")
+        print(f"Recommended fields: {analysis_result.get('recommended_fields', [])}")
+    except Exception as e:
+        print(f"❌ Analysis failed: {e}")
+        import traceback
+        traceback.print_exc()
+
+if __name__ == "__main__":
+    test_analysis_functions()