execution_programs.py

import io
import contextlib
import streamlit as st
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
from datetime import datetime
import re
from io import StringIO
import plotly.graph_objects as go
from llm_models import text_llm
from prompt_models import prompt_analyze


def analyze_code_execution(code: str, execution_result: dict, df: pd.DataFrame) -> str:
    """

    Analyze only the execution results (output and variables) to provide focused insights.

    

    Args:

        code: The Python code that was executed

        execution_result: The dictionary containing execution results

        df: The DataFrame being analyzed (for context only)

        

    Returns:

        str: Natural language analysis focused on the execution results

    """
    # Extract relevant information from execution_result
    output = execution_result.get('stdout', 'Tidak ada output teks')
    variables = execution_result.get('variables', {})
    error = execution_result.get('error', None)
    
    # Prepare variable summaries
    var_summaries = []
    for var_name, var_value in variables.items():
        if isinstance(var_value, (pd.DataFrame, pd.Series)):
            var_summaries.append(f"- {var_name}: {type(var_value).__name__} dengan bentuk {var_value.shape}")
        elif isinstance(var_value, (plt.Figure, sns.axisgrid.Grid)):
            var_summaries.append(f"- {var_name}: Visualisasi plot")
        else:
            var_summaries.append(f"- {var_name}: {type(var_value).__name__}")


    prompt = prompt_analyze(output, var_summaries, df, error)
    # Get analysis from the text model
    response = text_llm(prompt)

    # Clear thinking output
    clean_response = re.sub(r'<\s*think\s*>.*?<\s*/\s*think\s*>', '', response, flags=re.DOTALL | re.IGNORECASE)
    
    return clean_response

# =======================
def execute_code(code: str, df: pd.DataFrame):
    """Execute Python code safely and return outputs"""
    # Extract Python code from markdown blocks
    if "```python" in code:
        clean_code = code.split("```python")[1].split("```")[0].strip()
    else:
        clean_code = code.strip()
    
    # Add required imports if missing
    required_imports = [
        "import pandas as pd",
        "import matplotlib.pyplot as plt",
        "import seaborn as sns",
        "import plotly.express as px",
        "import plotly.graph_objects as go"
    ]
    
    for imp in required_imports:
        if imp not in clean_code:
            clean_code = imp + "\n" + clean_code
    
    # Prepare execution environment
    local_vars = {
        'df': df,
        'plt': plt,
        'sns': sns,
        'pd': pd
    }
    
    # Capture outputs
    stdout = io.StringIO()
    stderr = io.StringIO()
    
    try:
        with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr):
            exec(clean_code, local_vars)
            
                # Deteksi plot Plotly
            plotly_figs = []
            for var_name, var_value in local_vars.items():
                if isinstance(var_value, go.Figure):
                    plotly_figs.append(var_value)
            
            return {
                'code': clean_code,
                'stdout': stdout.getvalue(),
                'stderr': stderr.getvalue(),
                'figure': plotly_figs,
                'success': True,
                'variables': {k: v for k, v in local_vars.items() 
                              if not k.startswith('_') and k not in ['df', 'plt', 'sns', 'pd']}
            }

    except Exception as e:
        return {
            'code': clean_code,
            'error': str(e),
            'stderr': stderr.getvalue(),
            'success': False
        }

def add_to_history(execution_result):
    """Add execution result to history with timestamp"""
    history_item = {
        **execution_result,
        'timestamp': datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    }
    st.session_state.execution_history.append(history_item)

    # Generate explanation for successful executions
    if execution_result['success']:
        with st.spinner("Analyzing results..."):
            df_to_analyze = st.session_state.df_cleaned if 'df_cleaned' in st.session_state else st.session_state.df
            
            analysis = analyze_code_execution(
                code=execution_result['code'],
                execution_result=execution_result,
                df=df_to_analyze  # Only used for context, not analyzed
            )
            
            history_item['explanation'] = f"""

            **Analisis Hasil Eksekusi:**

            {analysis}

            """
            


# =======================
# Display Functions
# =======================

def display_code_with_highlight(code: str):
    """Display formatted Python code using Streamlit's built-in code block"""
    st.code(code, language='python')

def display_history():
    """Display all execution history items"""
    st.markdown("### Execution History")
    
    if not st.session_state.execution_history:
        st.info("No executions yet. Run some code to see results here.")
        return
    
    for i, item in enumerate(reversed(st.session_state.execution_history)):
        with st.container():
            st.markdown(f"### Execution #{len(st.session_state.execution_history)-i}")
            st.caption(f"Executed at {item['timestamp']}")
            
            # Display code
            with st.expander("View Code", expanded=False):
                display_code_with_highlight(item['code'])
            
            # Display outputs
            if item['success']:
                if 'explanation' in item:
                    with st.expander("See Explanation"):
                        st.markdown("**Explanation:**")
                        st.info(item['explanation'])

                if item['stdout']:
                    with st.expander("View Output", expanded=False):
                        st.markdown("**Output:**")
                        lines = item['stdout'].split('\n')
                        text_content = []
                        table_data = []
                        header_detected = False

                        for idx, line in enumerate(lines):
                            cleaned_line = line.strip()
                            if not cleaned_line:
                                continue
                            
                            if not header_detected:
                                if re.match(r"^[\w\s_]+$", cleaned_line) and len(cleaned_line.split()) > 1:
                                    if (idx + 1 < len(lines)) and re.match(r"^\d+\s+[\d\.e+-]+", lines[idx+1].strip()):
                                        header_detected = True
                                        table_data = lines[idx:]
                                        break
                                    else:
                                        text_content.append(cleaned_line)
                                else:
                                    text_content.append(cleaned_line)
                        
                        if text_content:
                            st.write("\n".join(text_content))
                            
                        if table_data:
                            try:
                                clean_table = [line.strip() for line in table_data if line.strip()]
                                df = pd.read_csv(StringIO("\n".join(clean_table)), sep=r"\s+", engine="python")
                                st.dataframe(df)
                            except Exception as e:
                                st.write("\n".join(table_data))

                
                if item['figure']:
                    st.markdown("**Visualisasi Interaktif:**")
                    for fig in item['figure']:
                        st.plotly_chart(fig, use_container_width=True)

                        
                if item['variables']:
                    with st.expander("Created Variables"):
                        st.json({k: str(type(v)) for k, v in item['variables'].items()})
            else:
                st.error("Execution failed")
                st.error(item['error'])
                if item['stderr']:
                    st.text("Error details:")
                    st.text(item['stderr'])
            
            st.markdown("---")