Reel_Analyzer / deploy.py
Agrannya's picture
Upload folder using huggingface_hub
6f0bbbb verified
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()