Spaces:
Runtime error
Runtime error
def plot_sentiment_pie(results, title="Reels Sentiment Analysis"): | |
""" | |
Creates a pie chart from sentiment analysis results and returns the matplotlib figure. | |
Args: | |
results: Counter object or dict with 'positive', 'neutral', 'negative' keys | |
title: Chart title | |
Returns: | |
Matplotlib Figure object, or None if no data. | |
""" | |
labels = ['Positive', 'Neutral', 'Negative'] | |
sizes = [results.get('positive', 0), results.get('neutral', 0), results.get('negative', 0)] | |
if sum(sizes) == 0: | |
return None | |
colors = ['#4CAF50', '#FFC107', '#F44336'] | |
explode = (0.05, 0, 0.05) | |
fig, ax = plt.subplots(figsize=(8, 6)) | |
filtered_labels = [label for i, label in enumerate(labels) if sizes[i] > 0] | |
filtered_sizes = [size for size in sizes if size > 0] | |
filtered_colors = [colors[i] for i, size in enumerate(sizes) if size > 0] | |
explode_map = {'Positive': 0.05, 'Neutral': 0, 'Negative': 0.05} | |
filtered_explode = [explode_map.get(label, 0) for label in filtered_labels] | |
ax.pie(filtered_sizes, explode=filtered_explode, labels=filtered_labels, colors=filtered_colors, | |
autopct='%1.1f%%', shadow=True, startangle=140, | |
textprops={'fontsize': 12, 'color': 'black'}) | |
ax.axis('equal') | |
plt.title(title, fontsize=16, pad=20) | |
plt.tight_layout() | |
# Return the figure object instead of saving to bytes | |
return fig | |
def plot_category_distribution(counter, title="Reels Content Distribution"): | |
""" | |
Generate pie chart from category counts and returns the matplotlib figure. | |
Args: | |
counter: Counter object with category counts. | |
title: Chart title. | |
Returns: | |
Matplotlib Figure object, or None if no data. | |
""" | |
labels = [] | |
sizes = [] | |
total = sum(counter.values()) | |
if total == 0: | |
return None | |
threshold = total * 0.02 | |
other_count = 0 | |
sorted_categories = counter.most_common() | |
for category, count in sorted_categories: | |
if count >= threshold and category != "other": | |
labels.append(category.replace('_', ' ').title()) | |
sizes.append(count) | |
elif category == "other": | |
other_count += count | |
else: | |
other_count += count | |
if other_count > 0: | |
labels.append("Other") | |
sizes.append(other_count) | |
if not sizes: | |
return None | |
fig, ax = plt.subplots(figsize=(10, 8)) | |
colors = plt.cm.viridis(np.linspace(0, 1, len(sizes))) | |
ax.pie( | |
sizes, | |
labels=labels, | |
autopct='%1.1f%%', | |
startangle=140, | |
colors=colors, | |
wedgeprops={'edgecolor': 'white', 'linewidth': 1}, | |
textprops={'fontsize': 11, 'color': 'black'} | |
) | |
plt.title(title, pad=20, fontsize=15) | |
plt.axis('equal') | |
plt.tight_layout() | |
# Return the figure object instead of saving to bytes | |
return fig | |
# The rest of the Gradio Blocks interface definition and function linking | |
# should remain the same, as the analyze_reels_gradio function already | |
# calls these plotting functions and is intended to return the figure objects now. | |
# Global variables to maintain state across Gradio calls | |
global cl | |
global explore_reels_list | |
global sentiment_analyzer_instance | |
global content_classifier_pipeline | |
# Initialize sentiment analyzer if not already done (can be done here or lazily in analyze_reels_gradio) | |
# Doing it here ensures the model is loaded when this cell runs, potentially reducing latency on first analyze click. | |
try: | |
sentiment_analyzer_instance = ReelSentimentAnalyzer() | |
print("Sentiment Analyzer initialized.") | |
# Optional: Train Hindi model if needed and data is available | |
# sample_train_data = [...] # Define your training data | |
# sentiment_analyzer_instance.train_hindi_model(sample_train_data) | |
except Exception as e: | |
print(f"Error initializing Sentiment Analyzer globally: {e}") | |
sentiment_analyzer_instance = None | |
# Initialize content classifier pipeline if not already done (can be done here or lazily) | |
try: | |
print("Initializing Content Classifier Pipeline globally...") | |
content_classifier_pipeline = pipeline( | |
"zero-shot-classification", | |
model="facebook/bart-large-mnli", | |
device=0 if torch.cuda.is_available() else -1 # Use GPU if available | |
) | |
print("Content Classifier Pipeline Initialized.") | |
except Exception as e: | |
print(f"Error initializing Content Classifier globally: {e}") | |
content_classifier_pipeline = None | |
def login_gradio(username): | |
"""Gradio-compatible login function.""" | |
global cl | |
try: | |
PASSWORD = userdata.get('password') | |
except Exception as e: | |
return f"Error accessing password secret: {e}" | |
if not PASSWORD: | |
return "Error: Instagram password not found in Colab secrets." | |
cl = Client() | |
try: | |
cl.login(username, PASSWORD) | |
return f"Successfully logged in as {username}" | |
except Exception as e: | |
cl = None # Ensure cl is None on failure | |
return f"Error during login: {e}" | |
def fetch_reels_gradio(): | |
"""Gradio-compatible function to fetch explore reels.""" | |
global cl | |
global explore_reels_list | |
if cl is None: | |
explore_reels_list = [] # Ensure list is empty on failure | |
return "Error: Not logged in. Please log in first." | |
try: | |
# Fetch a limited number of reels for demonstration purposes | |
# You might want to make this number configurable later | |
fetched_reels = cl.explore_reels()[:100] # Fetch up to 100 for analysis | |
explore_reels_list = fetched_reels | |
if explore_reels_list: | |
return f"Successfully fetched {len(explore_reels_list)} explore reels." | |
else: | |
explore_reels_list = [] # Ensure it's an empty list | |
return "Fetched 0 explore reels." | |
except Exception as e: | |
explore_reels_list = [] # Ensure it's an empty list on error | |
return f"Error fetching explore reels: {e}" | |
def analyze_reels_gradio(max_to_analyze): | |
"""Gradio-compatible function to analyze fetched reels and generate plots.""" | |
global explore_reels_list | |
global sentiment_analyzer_instance | |
global content_classifier_pipeline | |
if not explore_reels_list: | |
# Return None for plots if no reels | |
return "Error: No reels fetched yet. Please fetch reels first.", None, None | |
# Ensure max_to_analyze does not exceed the number of fetched reels | |
num_reels_to_process = min(max_to_analyze, len(explore_reels_list)) | |
reels_to_analyze = explore_reels_list[:num_reels_to_process] | |
if not reels_to_analyze: | |
return "Error: No reels available to analyze.", None, None | |
# Check if analyzers are initialized | |
if sentiment_analyzer_instance is None: | |
return "Error: Sentiment Analyzer not initialized.", None, None | |
if content_classifier_pipeline is None: | |
return "Error: Content Classifier not initialized.", None, None | |
analysis_status_messages = [] | |
sentiment_plot_figure = None # Changed to figure | |
content_plot_figure = None # Changed to figure | |
# Perform Sentiment Analysis | |
try: | |
analysis_status_messages.append(f"Starting Sentiment Analysis for {len(reels_to_analyze)} reels...") | |
sentiment_results, detailed_sentiment_results = sentiment_analyzer_instance.analyze_reels( | |
reels_to_analyze, | |
max_to_analyze=len(reels_to_analyze) # Pass the actual number being processed | |
) | |
# Call the updated plotting function that returns a figure | |
sentiment_plot_figure = plot_sentiment_pie(sentiment_results, title=f"Sentiment of {len(reels_to_analyze)} Instagram Reels") | |
analysis_status_messages.append("Sentiment Analysis Complete.") | |
except Exception as e: | |
analysis_status_messages.append(f"Error during Sentiment Analysis: {e}") | |
sentiment_plot_figure = None # Ensure plot is None on error | |
# Perform Content Categorization | |
try: | |
analysis_status_messages.append(f"Starting Content Categorization for {len(reels_to_analyze)} reels...") | |
category_counts = Counter() | |
# Re-implement content analysis slightly to fit this flow using the global pipeline | |
print(f"\n⏳ Analyzing content for {len(reels_to_analyze)} reels...") | |
for i, reel in enumerate(reels_to_analyze, 1): | |
caption = getattr(reel, 'caption_text', '') or getattr(reel, 'caption', '') or '' | |
# Use the global classifier pipeline | |
category, details = classify_reel_content(caption) | |
category_counts[category] += 1 | |
print("\n✅ Content Analysis complete!") | |
print("\n📊 Category Counts:") | |
for category, count in category_counts.most_common(): | |
print(f"- {category.replace('_', ' ').title()}: {count}") | |
# Call the updated plotting function that returns a figure | |
content_plot_figure = plot_category_distribution(category_counts) | |
analysis_status_messages.append("Content Categorization Complete.") | |
except Exception as e: | |
analysis_status_messages.append(f"Error during Content Analysis: {e}") | |
content_plot_figure = None # Ensure plot is None on error | |
final_status_message = "\n".join(analysis_status_messages) | |
# Return the figure objects | |
return final_status_message, sentiment_plot_figure, content_plot_figure | |
# --- Gradio Blocks Interface --- | |
with gr.Blocks() as demo: | |
gr.Markdown("# Instagram Reels Analysis") | |
with gr.Row(): | |
username_input = gr.Textbox(label="Instagram Username") | |
login_button = gr.Button("Login") | |
login_status_output = gr.Label(label="Login Status") | |
with gr.Row(): | |
fetch_button = gr.Button("Fetch Reels") | |
fetch_status_output = gr.Label(label="Fetch Status") | |
with gr.Row(): | |
max_reels_input = gr.Slider(minimum=1, maximum=100, value=10, step=1, label="Number of Reels to Analyze") | |
analyze_button = gr.Button("Analyze Reels") | |
analyze_status_output = gr.Label(label="Analysis Status") | |
with gr.Row(): | |
# Sentiment Analysis Outputs | |
with gr.Column(): | |
gr.Markdown("## Sentiment Analysis") | |
sentiment_plot_output = gr.Plot(label="Sentiment Distribution") | |
# Content Analysis Outputs | |
with gr.Column(): | |
gr.Markdown("## Content Analysis") | |
content_plot_output = gr.Plot(label="Content Distribution") | |
# Link login and fetch buttons (assuming login_gradio and fetch_reels_gradio are defined) | |
# Redefine login_gradio and fetch_reels_gradio here within the Blocks context | |
# to ensure they are linked correctly, even though they were defined above. | |
# This is a common pattern in Gradio Blocks. | |
def login_gradio_blocks(username): | |
"""Gradio-compatible login function for Blocks.""" | |
global cl | |
try: | |
PASSWORD = userdata.get('password') | |
except Exception as e: | |
return f"Error accessing password secret: {e}" | |
if not PASSWORD: | |
return "Error: Instagram password not found in Colab secrets." | |
cl = Client() | |
try: | |
cl.login(username, PASSWORD) | |
return f"Successfully logged in as {username}" | |
except Exception as e: | |
cl = None # Ensure cl is None on failure | |
return f"Error during login: {e}" | |
def fetch_reels_gradio_blocks(): | |
"""Gradio-compatible function to fetch explore reels for Blocks.""" | |
global cl | |
global explore_reels_list | |
if cl is None: | |
explore_reels_list = [] # Ensure list is empty on failure | |
return "Error: Not logged in. Please log in first." | |
try: | |
# Fetch a limited number of reels for demonstration purposes | |
# You might want to make this number configurable later | |
fetched_reels = cl.explore_reels()[:100] # Fetch up to 100 for analysis | |
explore_reels_list = fetched_reels | |
if explore_reels_list: | |
return f"Successfully fetched {len(explore_reels_list)} explore reels." | |
else: | |
explore_reels_list = [] # Ensure it's an empty list | |
return "Fetched 0 explore reels." | |
except Exception as e: | |
explore_reels_list = [] # Ensure it's an empty list on error | |
return f"Error fetching explore reels: {e}" | |
login_button.click( | |
fn=login_gradio_blocks, | |
inputs=username_input, | |
outputs=login_status_output | |
) | |
fetch_button.click( | |
fn=fetch_reels_gradio_blocks, | |
inputs=None, # No direct inputs needed for fetching | |
outputs=fetch_status_output | |
) | |
# Link the Analyze button to the analysis function | |
analyze_button.click( | |
fn=analyze_reels_gradio, | |
inputs=max_reels_input, # Input is the slider value | |
outputs=[analyze_status_output, sentiment_plot_output, content_plot_output] # Outputs are status and the two plots | |
) | |
# The demo is now fully defined. It can be launched in the next step. | |
# demo.launch() | |