Spaces:
Sleeping
Sleeping
| #18-08-05 | |
| # -*- coding: utf-8 -*- | |
| """BinaryMatrix_Production.ipynb | |
| Automatically generated by Colab. | |
| Original file is located at | |
| https://colab.research.google.com/drive/14s_B8s3ANig3-51wDw2PpBXnCCmBnbRo | |
| """ | |
| import gradio as gr | |
| import pandas as pd | |
| import numpy as np | |
| import os | |
| import re | |
| import tempfile | |
| import shutil | |
| from datetime import datetime | |
| from typing import List, Tuple, Dict, Any | |
| import json | |
| from io import BytesIO | |
| class CSVBinaryTransformer: | |
| def __init__(self): | |
| self.df = None | |
| self.original_filename = None | |
| self.processed_df = None | |
| def load_csv(self, file_path: str) -> Tuple[pd.DataFrame, List[str], str]: | |
| """Load CSV file and return dataframe, column list, and default topics column""" | |
| try: | |
| self.df = pd.read_csv(file_path) | |
| self.original_filename = os.path.splitext(os.path.basename(file_path))[0] | |
| # Create checkbox options for columns | |
| column_choices = [(col, col) for col in self.df.columns.tolist()] | |
| # Find default topics column (column before the one starting with "Categories") | |
| default_topics_column = "" | |
| columns = self.df.columns.tolist() | |
| for i, col in enumerate(columns): | |
| if col.startswith('Categories') and i > 0: | |
| default_topics_column = columns[i-1] | |
| break | |
| return self.df.head(10), column_choices, default_topics_column | |
| except Exception as e: | |
| raise gr.Error(f"Error loading CSV: {str(e)}") | |
| def select_all_columns(self) -> List[str]: | |
| """Return all column names for select all functionality""" | |
| if self.df is not None: | |
| return self.df.columns.tolist() | |
| return [] | |
| def deselect_all_columns(self) -> List[str]: | |
| """Return empty list for deselect all functionality""" | |
| return [] | |
| def process_absa_columns(self, df: pd.DataFrame) -> pd.DataFrame: | |
| """Process ABSA prefixed columns to create sentiment and topic columns""" | |
| absa_columns = [col for col in df.columns if col.startswith('ABSA')] | |
| if not absa_columns: | |
| return df | |
| # Create verbatim sentiment columns | |
| df['Verbatim_Positive'] = 0 | |
| df['Verbatim_Neutral'] = 0 | |
| df['Verbatim_Negative'] = 0 | |
| # Dictionary to store topic-sentiment combinations | |
| topic_sentiment_combinations = set() | |
| for col in absa_columns: | |
| for idx, value in df[col].items(): | |
| if pd.isna(value): | |
| continue | |
| value_str = str(value) | |
| # Split by pipe delimiter to get individual topic::sentiment pairs | |
| pairs = [pair.strip() for pair in value_str.split('|') if pair.strip()] | |
| for pair in pairs: | |
| if '::' in pair: | |
| topic_part, sentiment_part = pair.split('::', 1) | |
| topic = topic_part.strip() | |
| sentiment = sentiment_part.strip() | |
| if topic and sentiment: | |
| topic_sentiment_combinations.add((topic, sentiment)) | |
| # Update verbatim sentiment columns based on sentiment | |
| sentiment_lower = sentiment.lower() | |
| if 'positive' in sentiment_lower: | |
| df.at[idx, 'Verbatim_Positive'] = 1 | |
| elif 'negative' in sentiment_lower: | |
| df.at[idx, 'Verbatim_Negative'] = 1 | |
| elif 'neutral' in sentiment_lower: | |
| df.at[idx, 'Verbatim_Neutral'] = 1 | |
| # Create columns for topic-sentiment combinations | |
| for topic, sentiment in topic_sentiment_combinations: | |
| # Clean topic name for column naming | |
| safe_topic = re.sub(r'[^\w]', '_', topic).strip('_') | |
| col_name = f"S_{safe_topic}" | |
| if col_name not in df.columns: | |
| df[col_name] = "" | |
| # Fill the S_ columns with sentiment values | |
| for col in absa_columns: | |
| for idx, value in df[col].items(): | |
| if pd.isna(value): | |
| continue | |
| value_str = str(value) | |
| # Split by pipe delimiter to get individual topic::sentiment pairs | |
| pairs = [pair.strip() for pair in value_str.split('|') if pair.strip()] | |
| for pair in pairs: | |
| if '::' in pair: | |
| topic_part, sentiment_part = pair.split('::', 1) | |
| topic = topic_part.strip() | |
| sentiment = sentiment_part.strip() | |
| if topic and sentiment: | |
| # Clean topic name for column naming | |
| safe_topic = re.sub(r'[^\w]', '_', topic).strip('_') | |
| col_name = f"S_{safe_topic}" | |
| if col_name in df.columns: | |
| df.at[idx, col_name] = sentiment | |
| return df | |
| def process_categories_columns(self, df: pd.DataFrame) -> pd.DataFrame: | |
| """Process Categories prefixed columns to create binary category columns""" | |
| category_columns = [col for col in df.columns if col.startswith('Categories')] | |
| if not category_columns: | |
| return df | |
| # Collect all unique categories | |
| all_categories = set() | |
| for col in category_columns: | |
| for value in df[col].dropna(): | |
| if isinstance(value, str): | |
| # Split by common delimiters | |
| categories = re.split(r'[,;|]', value) | |
| for cat in categories: | |
| cat = cat.strip() | |
| if cat: | |
| all_categories.add(cat) | |
| # Create binary columns for each category | |
| category_columns_created = [] | |
| for category in all_categories: | |
| col_name = f"C_{category}" | |
| df[col_name] = 0 | |
| category_columns_created.append(col_name) | |
| for col in category_columns: | |
| for idx, value in df[col].items(): | |
| if pd.isna(value): | |
| continue | |
| if isinstance(value, str) and category.lower() in value.lower(): | |
| df.at[idx, col_name] = 1 | |
| # Sort C_ columns alphabetically by the text after C_ | |
| if category_columns_created: | |
| category_columns_created.sort(key=lambda x: x[2:]) # Sort by text after "C_" | |
| # Reorder the columns in the dataframe | |
| other_cols = [col for col in df.columns if not col.startswith('C_') or col not in category_columns_created] | |
| df = df[other_cols + category_columns_created] | |
| return df | |
| def process_topics_column(self, df: pd.DataFrame, topics_column: str) -> pd.DataFrame: | |
| """Process specified topics column to create binary topic columns""" | |
| if not topics_column or topics_column not in df.columns: | |
| return df | |
| # Collect all unique topics | |
| all_topics = set() | |
| for value in df[topics_column].dropna(): | |
| if isinstance(value, str): | |
| # Split only by pipe delimiter (|) to preserve commas within topics | |
| topics = [topic.strip() for topic in value.split('|') if topic.strip()] | |
| for topic in topics: | |
| if topic: | |
| all_topics.add(topic) | |
| # Create binary columns for each topic | |
| topic_columns_created = [] | |
| for topic in all_topics: | |
| col_name = f"T_{topic}" | |
| df[col_name] = 0 | |
| topic_columns_created.append(col_name) | |
| for idx, value in df[topics_column].items(): | |
| if pd.isna(value): | |
| continue | |
| if isinstance(value, str): | |
| # Split by pipe delimiter and check for exact match | |
| topics_in_row = [t.strip() for t in value.split('|')] | |
| if topic in topics_in_row: | |
| df.at[idx, col_name] = 1 | |
| # Sort T_ columns alphabetically by the text after T_ | |
| if topic_columns_created: | |
| topic_columns_created.sort(key=lambda x: x[2:]) # Sort by text after "T_" | |
| # Reorder the columns in the dataframe | |
| other_cols = [col for col in df.columns if not col.startswith('T_') or col not in topic_columns_created] | |
| df = df[other_cols + topic_columns_created] | |
| return df | |
| def save_transformed_data(self, output_format='xlsx'): | |
| """Save the transformed data and return the file path - FIXED Excel export""" | |
| if self.processed_df is None: | |
| raise ValueError("No transformed data to save") | |
| # Create filename with original filename prefix and timestamp | |
| timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") | |
| prefix = self.original_filename if self.original_filename else 'transformed_data' | |
| if output_format.lower() in ['xlsx', 'excel', 'excel (.xlsx)']: | |
| filename = f"{prefix}_BinaryTransformation_{timestamp}.xlsx" | |
| temp_file = tempfile.NamedTemporaryFile(delete=False, suffix='.xlsx') | |
| temp_file.close() # Close the file handle before writing | |
| # Use pandas to_excel with proper engine | |
| with pd.ExcelWriter(temp_file.name, engine='openpyxl') as writer: | |
| self.processed_df.to_excel(writer, index=False) | |
| final_path = os.path.join(tempfile.gettempdir(), filename) | |
| if os.path.exists(final_path): | |
| os.remove(final_path) | |
| os.rename(temp_file.name, final_path) | |
| else: # csv | |
| filename = f"{prefix}_BinaryTransformation_{timestamp}.csv" | |
| temp_file = tempfile.NamedTemporaryFile(delete=False, suffix='.csv') | |
| temp_file.close() # Close the file handle before writing | |
| self.processed_df.to_csv(temp_file.name, index=False) | |
| final_path = os.path.join(tempfile.gettempdir(), filename) | |
| if os.path.exists(final_path): | |
| os.remove(final_path) | |
| os.rename(temp_file.name, final_path) | |
| if not os.path.exists(final_path): | |
| raise ValueError(f"Failed to create output file: {final_path}") | |
| return final_path | |
| # Initialize transformer | |
| transformer = CSVBinaryTransformer() | |
| def handle_file_upload(file): | |
| """Handle CSV file upload - UPDATED to return default topics column and preselect all columns""" | |
| if file is None: | |
| return None, gr.update(choices=[], value=[]), "", "Please upload a CSV file" | |
| try: | |
| preview_df, column_choices, default_topics_column = transformer.load_csv(file.name) | |
| preview_html = preview_df.to_html(classes="table table-striped", table_id="upload-preview") | |
| # Preselect all columns | |
| all_columns = [col for col, _ in column_choices] | |
| return ( | |
| preview_html, | |
| gr.update(choices=column_choices, value=all_columns, visible=True), # Preselect all | |
| default_topics_column, # Set default topics column | |
| f"✅ Successfully loaded CSV with {len(transformer.df)} rows and {len(transformer.df.columns)} columns" | |
| ) | |
| except Exception as e: | |
| return None, gr.update(choices=[], value=[], visible=False), "", f"❌ Error: {str(e)}" | |
| def select_all(): | |
| """Select all columns""" | |
| if transformer.df is not None: | |
| return gr.update(value=transformer.select_all_columns()) | |
| return gr.update(value=[]) | |
| def deselect_all(): | |
| """Deselect all columns""" | |
| return gr.update(value=transformer.deselect_all_columns()) | |
| def process_transformation(selected_columns, topics_column, export_format): | |
| """Process the transformation - UPDATED to validate topics column""" | |
| try: | |
| if transformer.df is None: | |
| return None, None, "❌ Error: No CSV file loaded" | |
| if not selected_columns: | |
| return None, None, "❌ Error: Please select at least one column" | |
| # Validate topics column is provided and exists | |
| if not topics_column or topics_column.strip() == "": | |
| return None, None, "❌ Error: Topics Column Name is mandatory. Please specify a topics column." | |
| if topics_column not in transformer.df.columns: | |
| return None, None, f"❌ Error: Topics column '{topics_column}' not found in the data. Available columns: {', '.join(transformer.df.columns.tolist())}" | |
| # Create a copy of the dataframe with selected columns | |
| processed_df = transformer.df[selected_columns].copy() | |
| # Process ABSA columns | |
| processed_df = transformer.process_absa_columns(processed_df) | |
| # Process Categories columns | |
| processed_df = transformer.process_categories_columns(processed_df) | |
| # Process Topics column | |
| processed_df = transformer.process_topics_column(processed_df, topics_column) | |
| # Store processed data | |
| transformer.processed_df = processed_df | |
| # Generate preview | |
| preview_html = processed_df.head(20).to_html(classes="table table-striped", table_id="preview-table") | |
| # Save file with proper format detection | |
| format_clean = export_format.lower().replace(' (', '').replace(')', '').replace('.', '') | |
| output_file = transformer.save_transformed_data(format_clean) | |
| success_msg = f"✅ Transformation completed! Generated file: {os.path.basename(output_file)}" | |
| success_msg += f"\n📊 Processed {len(transformer.processed_df)} rows with {len(transformer.processed_df.columns)} columns" | |
| success_msg += f"\n💾 File saved successfully as {format_clean.upper()}" | |
| success_msg += f"\n📥 File download should start automatically" | |
| return preview_html, output_file, success_msg | |
| except Exception as e: | |
| import traceback | |
| error_msg = f"❌ Error during transformation: {str(e)}\n\n{traceback.format_exc()}" | |
| return None, None, error_msg | |
| # Create Gradio interface - UPDATED with colored buttons and mandatory field | |
| with gr.Blocks(title="CSV Binary Transformation Tool", theme=gr.themes.Soft()) as app: | |
| gr.Markdown(""" | |
| # 📊 CSV Binary Transformation Tool | |
| This tool transforms CSV files by creating binary columns for sentiment analysis, categories, and topics. | |
| ## Features: | |
| - **ABSA Processing**: Creates sentiment columns and topic-sentiment combinations | |
| - **Category Processing**: Creates binary columns for each category | |
| - **Topic Processing**: Creates binary columns for each topic (using "|" delimiter) | |
| - **Flexible Export**: Support for CSV and Excel formats | |
| """) | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| # File upload section | |
| gr.Markdown("### 1. Upload CSV File") | |
| file_input = gr.File( | |
| label="Upload Input File", | |
| file_types=[".csv"], | |
| type="filepath" | |
| ) | |
| upload_status = gr.Textbox( | |
| label="Upload Status", | |
| interactive=False, | |
| lines=2 | |
| ) | |
| # Column selection section | |
| gr.Markdown("### 2. Select Columns") | |
| gr.Markdown("*Choose which columns from your CSV to include in the output file*") | |
| with gr.Row(): | |
| # Colored buttons as requested | |
| select_all_btn = gr.Button("✓ Select All", size="sm", variant="primary", elem_classes="green-button") | |
| deselect_all_btn = gr.Button("✗ Deselect All", size="sm", variant="secondary", elem_classes="red-button") | |
| column_selector = gr.CheckboxGroup( | |
| label="Choose columns to include in output", | |
| choices=[], | |
| value=[], | |
| interactive=True, | |
| visible=False, | |
| info="Select the columns you want to include in the transformed output file" | |
| ) | |
| # Topics column input - MADE MANDATORY | |
| gr.Markdown("### 3. Specify Topics Column **(MANDATORY)**") | |
| gr.Markdown("⚠️ **This field is required** - Enter the name of the column containing topics") | |
| gr.Markdown("**Note**: Topics should be separated by '|' delimiter in your data") | |
| topics_column_input = gr.Textbox( | |
| label="Topics Column Name (Required)", | |
| placeholder="This field will be auto-populated when you upload a CSV", | |
| info="This column will be used to create T_<topic> binary columns (using '|' as delimiter)", | |
| interactive=True | |
| ) | |
| # Export options | |
| gr.Markdown("### 4. Export Settings") | |
| export_format = gr.Radio( | |
| label="Output Format", | |
| choices=["CSV (.csv)", "Excel (.xlsx)"], | |
| value="Excel (.xlsx)" | |
| ) | |
| # Process button | |
| process_btn = gr.Button( | |
| "🚀 Transform Data", | |
| variant="primary", | |
| size="lg" | |
| ) | |
| with gr.Column(scale=2): | |
| # Preview sections | |
| gr.Markdown("### File Preview") | |
| with gr.Tabs(): | |
| with gr.Tab("Original Data"): | |
| original_preview = gr.HTML( | |
| label="Original Data Preview (First 10 rows)", | |
| value="<p>No file uploaded yet</p>" | |
| ) | |
| with gr.Tab("Transformed Data"): | |
| transformed_preview = gr.HTML( | |
| label="Transformed Data Preview (First 20 rows)", | |
| value="<p>No transformation performed yet</p>" | |
| ) | |
| # Status and download | |
| process_status = gr.Textbox( | |
| label="Processing Status", | |
| interactive=False, | |
| lines=6 | |
| ) | |
| # Download section | |
| gr.Markdown("### 📥 Download Status") | |
| gr.Markdown("Please click on the link inside the output file size value to download the transformed file (the number value on the right hand side below). You may need to right click and select Save Link As (or something similar)") | |
| output_file = gr.File( | |
| label="Transformed Binary Matrix (Auto-Download)", | |
| interactive=False, | |
| visible=True | |
| ) | |
| # Event handlers - UPDATED to include topics column default | |
| file_input.change( | |
| fn=handle_file_upload, | |
| inputs=[file_input], | |
| outputs=[original_preview, column_selector, topics_column_input, upload_status] | |
| ) | |
| select_all_btn.click( | |
| fn=select_all, | |
| outputs=[column_selector] | |
| ) | |
| deselect_all_btn.click( | |
| fn=deselect_all, | |
| outputs=[column_selector] | |
| ) | |
| process_btn.click( | |
| fn=process_transformation, | |
| inputs=[column_selector, topics_column_input, export_format], | |
| outputs=[transformed_preview, output_file, process_status] | |
| ) | |
| # Add custom CSS for better styling and colored buttons | |
| app.load(js=""" | |
| function() { | |
| const style = document.createElement('style'); | |
| style.textContent = ` | |
| .table { font-size: 12px; } | |
| .table th, .table td { padding: 4px 8px; } | |
| #upload-preview, #preview-table { max-height: 400px; overflow-y: auto; } | |
| /* Green button for Select All */ | |
| .green-button button { | |
| background-color: #22c55e !important; | |
| border-color: #16a34a !important; | |
| color: white !important; | |
| } | |
| .green-button button:hover { | |
| background-color: #16a34a !important; | |
| border-color: #15803d !important; | |
| } | |
| /* Pale red button for Deselect All */ | |
| .red-button button { | |
| background-color: #fca5a5 !important; | |
| border-color: #f87171 !important; | |
| color: #7f1d1d !important; | |
| } | |
| .red-button button:hover { | |
| background-color: #f87171 !important; | |
| border-color: #ef4444 !important; | |
| color: #7f1d1d !important; | |
| } | |
| `; | |
| document.head.appendChild(style); | |
| } | |
| """) | |
| if __name__ == "__main__": | |
| app.launch( | |
| ) |