""" BI Storyteller CLI Interface Command-line interface for marketing analysis automation Standard Library Only - No Network Dependencies """ import json import os from main import BIStoryteller class BIStoryteller_CLI: """Command-line interface for BI Storyteller""" def __init__(self): self.bi = BIStoryteller() self.current_step = 1 def print_header(self): """Print application header""" print("\n" + "="*60) print("๐Ÿš€ BI STORYTELLER - MARKETING ANALYSIS PLATFORM") print("="*60) print("๐Ÿ“Š Complete workflow for marketing data analysis") print("๐Ÿ”ง Standard Library Only - No External Dependencies") print("="*60) def print_menu(self): """Print main menu""" print(f"\n๐Ÿ“‹ MAIN MENU (Current Step: {self.current_step}/12)") print("-" * 40) print("1. ๐Ÿ”‘ Set API Key (Optional)") print("2. ๐Ÿ“ Extract Variables") print("3. ๐Ÿ“‹ Generate Questionnaire") print("4. ๐Ÿ”ข Generate Sample Data") print("5. ๐Ÿงน Clean Data") print("6. ๐Ÿ“Š Perform EDA") print("7. ๐Ÿค– Train Predictive Model") print("8. ๐Ÿ“ˆ Analyze Trends") print("9. ๐Ÿ’ญ Analyze Sentiment") print("10. ๐Ÿงช Run A/B Test") print("11. ๐Ÿ’ฌ Chat with Data") print("12. ๐Ÿ“ค Export Results") print("-" * 40) print("13. ๐Ÿ“ฅ Import Previous Analysis") print("14. ๐Ÿ“„ Export Data as CSV") print("15. โŒ Exit") print("-" * 40) def get_user_input(self, prompt, input_type="string"): """Get user input with validation""" while True: try: user_input = input(f"\n{prompt}: ").strip() if input_type == "int": return int(user_input) elif input_type == "float": return float(user_input) else: return user_input except ValueError: print(f"โŒ Please enter a valid {input_type}") except KeyboardInterrupt: print("\n๐Ÿ‘‹ Goodbye!") exit(0) def print_results(self, title, results, success_key="success"): """Print formatted results""" print(f"\n{title}") print("-" * len(title)) if results.get(success_key): if "results" in results: self.print_dict(results["results"], indent=0) else: self.print_dict(results, indent=0) else: print(f"โŒ Error: {results.get('error', 'Unknown error')}") def print_dict(self, data, indent=0): """Print dictionary in a formatted way""" spaces = " " * indent for key, value in data.items(): if isinstance(value, dict): print(f"{spaces}{key}:") self.print_dict(value, indent + 1) elif isinstance(value, list): print(f"{spaces}{key}: [{len(value)} items]") if value and len(value) <= 5: for item in value: print(f"{spaces} โ€ข {item}") else: print(f"{spaces}{key}: {value}") def module_1_api_key(self): """Module 1: Set API Key""" print("\n๐Ÿ”‘ MODULE 1: API KEY SETUP") print("=" * 30) print("Enter your Groq API key for AI-powered analysis.") print("Leave empty to use offline mode with fallback functionality.") api_key = self.get_user_input("Groq API Key (or press Enter to skip)") if api_key: result = self.bi.set_groq_api_key(api_key) self.print_results("โœ… API Key Setup", result) else: print("โšก Using offline mode - fallback analysis will be used") self.current_step = max(self.current_step, 2) input("\nPress Enter to continue...") def module_2_extract_variables(self): """Module 2: Extract Variables""" print("\n๐Ÿ“ MODULE 2: VARIABLE EXTRACTION") print("=" * 35) print("Describe your business problem to extract relevant variables.") business_problem = self.get_user_input("Business Problem Description") if business_problem: result = self.bi.extract_variables(business_problem) self.print_results("โœ… Variable Extraction Results", result) self.current_step = max(self.current_step, 3) else: print("โŒ Please provide a business problem description") input("\nPress Enter to continue...") def module_3_generate_questionnaire(self): """Module 3: Generate Questionnaire""" print("\n๐Ÿ“‹ MODULE 3: QUESTIONNAIRE GENERATION") print("=" * 40) if not self.bi.variables: print("โŒ Please extract variables first (Module 2)") input("Press Enter to continue...") return result = self.bi.generate_questionnaire(self.bi.variables, "") self.print_results("โœ… Questionnaire Generation Results", result) if result.get("success"): print("\n๐Ÿ“ Sample Questions:") for i, question in enumerate(result["questionnaire"][:3]): print(f"{i+1}. {question['question']}") self.current_step = max(self.current_step, 4) input("\nPress Enter to continue...") def module_4_generate_data(self): """Module 4: Generate Sample Data""" print("\n๐Ÿ”ข MODULE 4: SAMPLE DATA GENERATION") print("=" * 38) if not self.bi.variables: print("โŒ Please extract variables first (Module 2)") input("Press Enter to continue...") return sample_size = self.get_user_input("Sample Size (100-10000)", "int") if 100 <= sample_size <= 10000: result = self.bi.generate_sample_data(self.bi.variables, sample_size) self.print_results("โœ… Sample Data Generation Results", result) if result.get("success"): print(f"\n๐Ÿ“Š Sample Record:") self.print_dict(result["data"][0], indent=1) self.current_step = max(self.current_step, 5) else: print("โŒ Sample size must be between 100 and 10,000") input("\nPress Enter to continue...") def module_5_clean_data(self): """Module 5: Clean Data""" print("\n๐Ÿงน MODULE 5: DATA CLEANING") print("=" * 28) if not self.bi.sample_data: print("โŒ Please generate sample data first (Module 4)") input("Press Enter to continue...") return result = self.bi.clean_data(self.bi.sample_data) self.print_results("โœ… Data Cleaning Results", result) self.current_step = max(self.current_step, 6) input("\nPress Enter to continue...") def module_6_perform_eda(self): """Module 6: Perform EDA""" print("\n๐Ÿ“Š MODULE 6: EXPLORATORY DATA ANALYSIS") print("=" * 40) if not self.bi.cleaned_data: print("โŒ Please clean data first (Module 5)") input("Press Enter to continue...") return result = self.bi.perform_eda(self.bi.cleaned_data) self.print_results("โœ… EDA Analysis Results", result) self.current_step = max(self.current_step, 7) input("\nPress Enter to continue...") def module_7_train_model(self): """Module 7: Train Predictive Model""" print("\n๐Ÿค– MODULE 7: PREDICTIVE ANALYTICS") print("=" * 35) if not self.bi.cleaned_data: print("โŒ Please clean data first (Module 5)") input("Press Enter to continue...") return print("Available algorithms:") algorithms = ["Random Forest", "Logistic Regression", "SVM", "Neural Network"] for i, alg in enumerate(algorithms, 1): print(f"{i}. {alg}") choice = self.get_user_input("Select algorithm (1-4)", "int") if 1 <= choice <= 4: algorithm = algorithms[choice - 1] result = self.bi.train_predictive_model(self.bi.cleaned_data, algorithm) self.print_results("โœ… Predictive Model Results", result) self.current_step = max(self.current_step, 8) else: print("โŒ Invalid algorithm selection") input("\nPress Enter to continue...") def module_8_analyze_trends(self): """Module 8: Analyze Trends""" print("\n๐Ÿ“ˆ MODULE 8: TREND ANALYSIS") print("=" * 28) if not self.bi.cleaned_data: print("โŒ Please clean data first (Module 5)") input("Press Enter to continue...") return print("Time periods:") periods = ["Daily", "Weekly", "Monthly"] for i, period in enumerate(periods, 1): print(f"{i}. {period}") choice = self.get_user_input("Select time period (1-3)", "int") if 1 <= choice <= 3: time_period = periods[choice - 1] result = self.bi.analyze_trends(self.bi.cleaned_data, time_period) self.print_results("โœ… Trend Analysis Results", result) self.current_step = max(self.current_step, 9) else: print("โŒ Invalid time period selection") input("\nPress Enter to continue...") def module_9_analyze_sentiment(self): """Module 9: Analyze Sentiment""" print("\n๐Ÿ’ญ MODULE 9: SENTIMENT ANALYSIS") print("=" * 32) if not self.bi.cleaned_data: print("โŒ Please clean data first (Module 5)") input("Press Enter to continue...") return result = self.bi.analyze_sentiment(self.bi.cleaned_data) self.print_results("โœ… Sentiment Analysis Results", result) self.current_step = max(self.current_step, 10) input("\nPress Enter to continue...") def module_10_ab_test(self): """Module 10: Run A/B Test""" print("\n๐Ÿงช MODULE 10: A/B TESTING") print("=" * 25) if not self.bi.cleaned_data: print("โŒ Please clean data first (Module 5)") input("Press Enter to continue...") return print("Available variables:") if self.bi.variables: for i, var in enumerate(self.bi.variables, 1): print(f"{i}. {var}") test_variable = self.get_user_input("Test Variable") success_metric = self.get_user_input("Success Metric") if test_variable and success_metric: result = self.bi.run_ab_test(self.bi.cleaned_data, test_variable, success_metric) self.print_results("โœ… A/B Test Results", result) self.current_step = max(self.current_step, 11) else: print("โŒ Please provide both test variable and success metric") input("\nPress Enter to continue...") def module_11_chat(self): """Module 11: Chat with Data""" print("\n๐Ÿ’ฌ MODULE 11: CHAT WITH DATA") print("=" * 30) print("Ask questions about your analysis. Type 'back' to return to menu.") while True: question = self.get_user_input("\nโ“ Your Question (or 'back' to exit)") if question.lower() == 'back': break result = self.bi.chat_with_data(question) if result.get("success"): print(f"\n๐Ÿค– Response: {result['response']}") print(f"๐Ÿ“Š Context Used: {result['context_used']} analysis modules") else: print(f"โŒ Error: {result.get('error', 'Unknown error')}") self.current_step = max(self.current_step, 12) def module_12_export(self): """Module 12: Export Results""" print("\n๐Ÿ“ค MODULE 12: EXPORT RESULTS") print("=" * 30) filename = self.get_user_input("Export filename (or press Enter for auto-generated)") if not filename: filename = None result = self.bi.export_results(filename) self.print_results("โœ… Export Results", result) def module_13_import(self): """Module 13: Import Previous Analysis""" print("\n๐Ÿ“ฅ IMPORT PREVIOUS ANALYSIS") print("=" * 30) # List available JSON files json_files = [f for f in os.listdir('.') if f.endswith('.json')] if json_files: print("Available analysis files:") for i, file in enumerate(json_files, 1): print(f"{i}. {file}") choice = self.get_user_input("Select file number", "int") if 1 <= choice <= len(json_files): filename = json_files[choice - 1] result = self.bi.import_results(filename) self.print_results("โœ… Import Results", result) if result.get("success"): self.current_step = 12 # Set to final step else: print("โŒ Invalid file selection") else: filename = self.get_user_input("Enter filename to import") result = self.bi.import_results(filename) self.print_results("โœ… Import Results", result) def module_14_export_csv(self): """Module 14: Export Data as CSV""" print("\n๐Ÿ“„ EXPORT DATA AS CSV") print("=" * 25) print("Data types:") print("1. Sample Data") print("2. Cleaned Data") choice = self.get_user_input("Select data type (1-2)", "int") if choice == 1: result = self.bi.export_data_csv("sample") elif choice == 2: result = self.bi.export_data_csv("cleaned") else: print("โŒ Invalid selection") return self.print_results("โœ… CSV Export Results", result) def run(self): """Main CLI loop""" self.print_header() while True: self.print_menu() try: choice = self.get_user_input("Select option (1-15)", "int") if choice == 1: self.module_1_api_key() elif choice == 2: self.module_2_extract_variables() elif choice == 3: self.module_3_generate_questionnaire() elif choice == 4: self.module_4_generate_data() elif choice == 5: self.module_5_clean_data() elif choice == 6: self.module_6_perform_eda() elif choice == 7: self.module_7_train_model() elif choice == 8: self.module_8_analyze_trends() elif choice == 9: self.module_9_analyze_sentiment() elif choice == 10: self.module_10_ab_test() elif choice == 11: self.module_11_chat() elif choice == 12: self.module_12_export() elif choice == 13: self.module_13_import() elif choice == 14: self.module_14_export_csv() elif choice == 15: print("\n๐Ÿ‘‹ Thank you for using BI Storyteller!") break else: print("โŒ Invalid option. Please select 1-15.") except KeyboardInterrupt: print("\n\n๐Ÿ‘‹ Goodbye!") break except Exception as e: print(f"โŒ An error occurred: {str(e)}") input("Press Enter to continue...") def module_1_api_key(self): """Module 1: Set API Key""" print("\n๐Ÿ”‘ MODULE 1: API KEY SETUP") print("=" * 30) print("Enter your Groq API key for AI-powered analysis.") print("Leave empty to use offline mode with fallback functionality.") api_key = self.get_user_input("Groq API Key (or press Enter to skip)") if api_key: result = self.bi.set_groq_api_key(api_key) self.print_results("โœ… API Key Setup", result) else: print("โšก Using offline mode - fallback analysis will be used") self.current_step = max(self.current_step, 2) input("\nPress Enter to continue...") def module_2_extract_variables(self): """Module 2: Extract Variables""" print("\n๐Ÿ“ MODULE 2: VARIABLE EXTRACTION") print("=" * 35) business_problem = self.get_user_input("Describe your business problem") if business_problem: result = self.bi.extract_variables(business_problem) self.print_results("โœ… Variable Extraction Results", result) if result.get("success"): print(f"\n๐Ÿ“Š Extracted Variables:") for var in result["variables"]: print(f" โ€ข {var.replace('_', ' ').title()}") self.current_step = max(self.current_step, 3) else: print("โŒ Please provide a business problem description") input("\nPress Enter to continue...") def module_3_generate_questionnaire(self): """Module 3: Generate Questionnaire""" print("\n๐Ÿ“‹ MODULE 3: QUESTIONNAIRE GENERATION") print("=" * 40) if not self.bi.variables: print("โŒ Please extract variables first (Module 2)") input("Press Enter to continue...") return result = self.bi.generate_questionnaire(self.bi.variables, "") self.print_results("โœ… Questionnaire Generation Results", result) if result.get("success"): print("\n๐Ÿ“ Sample Questions:") for i, question in enumerate(result["questionnaire"][:3]): print(f"{i+1}. {question['question']}") if question["type"] == "multiple_choice": print(f" Options: {', '.join(question['options'])}") self.current_step = max(self.current_step, 4) input("\nPress Enter to continue...") def module_4_generate_data(self): """Module 4: Generate Sample Data""" print("\n๐Ÿ”ข MODULE 4: SAMPLE DATA GENERATION") print("=" * 38) if not self.bi.variables: print("โŒ Please extract variables first (Module 2)") input("Press Enter to continue...") return sample_size = self.get_user_input("Sample Size (100-10000)", "int") if 100 <= sample_size <= 10000: print(f"๐Ÿ”„ Generating {sample_size} sample records...") result = self.bi.generate_sample_data(self.bi.variables, sample_size) self.print_results("โœ… Sample Data Generation Results", result) if result.get("success"): print(f"\n๐Ÿ“Š Sample Record:") sample_record = {k: v for k, v in result["data"][0].items() if k != "timestamp"} self.print_dict(sample_record, indent=1) self.current_step = max(self.current_step, 5) else: print("โŒ Sample size must be between 100 and 10,000") input("\nPress Enter to continue...") def module_5_clean_data(self): """Module 5: Clean Data""" print("\n๐Ÿงน MODULE 5: DATA CLEANING") print("=" * 28) if not self.bi.sample_data: print("โŒ Please generate sample data first (Module 4)") input("Press Enter to continue...") return print("๐Ÿ”„ Cleaning data...") result = self.bi.clean_data(self.bi.sample_data) if result.get("success"): print(f"โœ… Data cleaning completed!") print(f"๐Ÿ“Š Original records: {result['original_size']}") print(f"๐Ÿ“Š Cleaned records: {result['cleaned_size']}") print(f"๐Ÿ—‘๏ธ Outliers removed: {result['removed_outliers']}") print(f"๐Ÿ“ˆ Data quality: {((result['cleaned_size'] / result['original_size']) * 100):.1f}%") else: print(f"โŒ Error: {result.get('error', 'Unknown error')}") self.current_step = max(self.current_step, 6) input("\nPress Enter to continue...") def module_6_perform_eda(self): """Module 6: Perform EDA""" print("\n๐Ÿ“Š MODULE 6: EXPLORATORY DATA ANALYSIS") print("=" * 40) if not self.bi.cleaned_data: print("โŒ Please clean data first (Module 5)") input("Press Enter to continue...") return print("๐Ÿ”„ Performing exploratory data analysis...") result = self.bi.perform_eda(self.bi.cleaned_data) if result.get("success"): print("โœ… EDA Analysis completed!") # Show key insights if result["results"].get("insights"): print("\n๐Ÿ” Key Insights:") for insight in result["results"]["insights"]: print(f" โ€ข {insight}") # Show top correlations if result["results"].get("correlations"): print("\n๐Ÿ“ˆ Top Correlations:") correlations = sorted(result["results"]["correlations"].items(), key=lambda x: abs(x[1]), reverse=True)[:5] for pair, corr in correlations: print(f" โ€ข {pair}: {corr}") else: print(f"โŒ Error: {result.get('error', 'Unknown error')}") self.current_step = max(self.current_step, 7) input("\nPress Enter to continue...") def module_7_train_model(self): """Module 7: Train Predictive Model""" print("\n๐Ÿค– MODULE 7: PREDICTIVE ANALYTICS") print("=" * 35) if not self.bi.cleaned_data: print("โŒ Please clean data first (Module 5)") input("Press Enter to continue...") return print("Available algorithms:") algorithms = ["Random Forest", "Logistic Regression", "SVM", "Neural Network"] for i, alg in enumerate(algorithms, 1): print(f"{i}. {alg}") choice = self.get_user_input("Select algorithm (1-4)", "int") if 1 <= choice <= 4: algorithm = algorithms[choice - 1] print(f"๐Ÿ”„ Training {algorithm} model...") result = self.bi.train_predictive_model(self.bi.cleaned_data, algorithm) if result.get("success"): print(f"โœ… Model training completed!") print(f"๐ŸŽฏ Algorithm: {result['results']['algorithm']}") print(f"๐Ÿ“Š Accuracy: {(result['results']['metrics']['accuracy'] * 100):.1f}%") print(f"๐Ÿ“Š Precision: {(result['results']['metrics']['precision'] * 100):.1f}%") print(f"๐Ÿ“Š Recall: {(result['results']['metrics']['recall'] * 100):.1f}%") # Show feature importance if result["results"].get("feature_importance"): print("\n๐Ÿ” Top Feature Importance:") importance = sorted(result["results"]["feature_importance"].items(), key=lambda x: x[1], reverse=True)[:5] for feature, imp in importance: print(f" โ€ข {feature}: {(imp * 100):.1f}%") else: print(f"โŒ Error: {result.get('error', 'Unknown error')}") self.current_step = max(self.current_step, 8) else: print("โŒ Invalid algorithm selection") input("\nPress Enter to continue...") def module_8_analyze_trends(self): """Module 8: Analyze Trends""" print("\n๐Ÿ“ˆ MODULE 8: TREND ANALYSIS") print("=" * 28) if not self.bi.cleaned_data: print("โŒ Please clean data first (Module 5)") input("Press Enter to continue...") return print("Time periods:") periods = ["Daily", "Weekly", "Monthly"] for i, period in enumerate(periods, 1): print(f"{i}. {period}") choice = self.get_user_input("Select time period (1-3)", "int") if 1 <= choice <= 3: time_period = periods[choice - 1] print(f"๐Ÿ”„ Analyzing {time_period.lower()} trends...") result = self.bi.analyze_trends(self.bi.cleaned_data, time_period) if result.get("success"): print(f"โœ… Trend analysis completed!") print(f"๐Ÿ“Š Time Period: {result['results']['time_period']}") print(f"๐Ÿ“Š Analysis Periods: {result['results']['analysis_periods']}") # Show trends if result["results"].get("trends"): print("\n๐Ÿ“ˆ Key Trends:") for variable, trend in result["results"]["trends"].items(): print(f" โ€ข {variable}: {trend['direction']} (slope: {trend['slope']})") else: print(f"โŒ Error: {result.get('error', 'Unknown error')}") self.current_step = max(self.current_step, 9) else: print("โŒ Invalid time period selection") input("\nPress Enter to continue...") def module_9_analyze_sentiment(self): """Module 9: Analyze Sentiment""" print("\n๐Ÿ’ญ MODULE 9: SENTIMENT ANALYSIS") print("=" * 32) if not self.bi.cleaned_data: print("โŒ Please clean data first (Module 5)") input("Press Enter to continue...") return print("๐Ÿ”„ Analyzing sentiment...") result = self.bi.analyze_sentiment(self.bi.cleaned_data) if result.get("success"): print("โœ… Sentiment analysis completed!") print(f"๐Ÿ“Š Total Analyzed: {result['results']['total_analyzed']}") print(f"๐ŸŽฏ Dominant Sentiment: {result['results']['dominant_sentiment']}") print("\n๐Ÿ“Š Sentiment Distribution:") for sentiment, percentage in result["results"]["sentiment_distribution"].items(): print(f" โ€ข {sentiment}: {percentage}%") else: print(f"โŒ Error: {result.get('error', 'Unknown error')}") self.current_step = max(self.current_step, 10) input("\nPress Enter to continue...") def module_10_ab_test(self): """Module 10: Run A/B Test""" print("\n๐Ÿงช MODULE 10: A/B TESTING") print("=" * 25) if not self.bi.cleaned_data: print("โŒ Please clean data first (Module 5)") input("Press Enter to continue...") return print("Available variables:") if self.bi.variables: for i, var in enumerate(self.bi.variables, 1): print(f" {i}. {var}") test_variable = self.get_user_input("Test Variable") success_metric = self.get_user_input("Success Metric") if test_variable and success_metric: print("๐Ÿ”„ Running A/B test...") result = self.bi.run_ab_test(self.bi.cleaned_data, test_variable, success_metric) if result.get("success"): print("โœ… A/B test completed!") print(f"๐Ÿ‘ฅ Group A: {result['results']['group_a']['size']} users, {(result['results']['group_a']['success_rate'] * 100):.1f}% success") print(f"๐Ÿ‘ฅ Group B: {result['results']['group_b']['size']} users, {(result['results']['group_b']['success_rate'] * 100):.1f}% success") print(f"๐Ÿ“Š P-Value: {result['results']['statistical_test']['p_value']}") print(f"๐Ÿ† Winner: {result['results']['conclusion']['winner']}") print(f"๐Ÿ“ˆ Lift: {result['results']['conclusion']['lift']}%") else: print(f"โŒ Error: {result.get('error', 'Unknown error')}") self.current_step = max(self.current_step, 11) else: print("โŒ Please provide both test variable and success metric") input("\nPress Enter to continue...") def module_11_chat(self): """Module 11: Chat with Data""" print("\n๐Ÿ’ฌ MODULE 11: CHAT WITH DATA") print("=" * 30) print("Ask questions about your analysis. Type 'back' to return to menu.") while True: question = self.get_user_input("\nโ“ Your Question (or 'back' to exit)") if question.lower() == 'back': break result = self.bi.chat_with_data(question) if result.get("success"): print(f"\n๐Ÿค– Response: {result['response']}") print(f"๐Ÿ“Š Context Used: {result['context_used']} analysis modules") else: print(f"โŒ Error: {result.get('error', 'Unknown error')}") self.current_step = max(self.current_step, 12) def module_12_export(self): """Module 12: Export Results""" print("\n๐Ÿ“ค MODULE 12: EXPORT RESULTS") print("=" * 30) filename = self.get_user_input("Export filename (or press Enter for auto-generated)") if not filename: filename = None print("๐Ÿ”„ Exporting analysis results...") result = self.bi.export_results(filename) if result.get("success"): print("โœ… Export completed!") print(f"๐Ÿ“ Filename: {result['filename']}") print(f"๐Ÿ“Š Modules Completed: {result['modules_completed']}") print(f"๐Ÿ’พ File Size: {(result['file_size'] / 1024):.1f} KB") else: print(f"โŒ Error: {result.get('error', 'Unknown error')}") def main(): """Main function to start CLI interface""" cli = BIStoryteller_CLI() cli.run() if __name__ == "__main__": main()