Post_Extractor / app.py
ahm14's picture
Update app.py
2976ba1 verified
import gradio as gr
import requests
import pytesseract
from PIL import Image
import docx
from transformers import pipeline
from keybert import KeyBERT
from io import BytesIO
from langdetect import detect
import re
import asyncio
from twscrape import API, gather
from selenium.webdriver.chrome.options import Options
from selenium import webdriver
from webdriver_manager.chrome import ChromeDriverManager
from bs4 import BeautifulSoup
import time
# Set up Tesseract
# pytesseract.pytesseract.tesseract_cmd = r'C:\Program Files\Tesseract-OCR\tesseract.exe' # Uncomment for Windows
# Initialize AI models
emotion_classifier = pipeline("text-classification", model="joeddav/distilbert-base-uncased-go-emotions-student")
keyword_extractor = KeyBERT()
zero_shot_classifier = pipeline("zero-shot-classification", model="facebook/bart-large-mnli")
class RealTimeSocialScraper:
def __init__(self):
self.api = API() # Configure proxies if needed
self.driver = self._init_browser()
def _init_browser(self):
chrome_options = Options()
chrome_options.add_argument("--headless")
chrome_options.add_argument("--disable-gpu")
return webdriver.Chrome(ChromeDriverManager().install(), options=chrome_options)
async def scrape(self, platform, query, limit=10):
if platform == "twitter":
return await self._scrape_twitter(query, limit)
elif platform == "instagram":
return self._scrape_instagram(query)
elif platform == "tiktok":
return self._scrape_tiktok(query)
else:
raise ValueError(f"Unsupported platform: {platform}")
async def _scrape_twitter(self, query, limit):
await self.api.pool.login_all()
return await gather(self.api.search(query, limit=limit))
def _scrape_instagram(self, query):
self.driver.get(f"https://www.instagram.com/explore/tags/{query}/")
WebDriverWait(self.driver, 30).until(
EC.presence_of_element_located((By.CLASS_NAME, "v1Nh3"))
)
soup = BeautifulSoup(self.driver.page_source, 'html.parser')
posts = []
for post in soup.findAll("div", class_="v1Nh3"):
posts.append({
'content': post.find('img')['alt'],
'image_url': post.find('img')['src']
})
return posts[:10]
def _scrape_tiktok(self, query):
# Implement TikTok scraping logic or use API
return [{"content": f"Demo TikTok post about {query}"}]
async def extract_posts(profile_url, hashtags, num_posts):
scraper = RealTimeSocialScraper()
platform = "twitter" if "twitter" in profile_url else "instagram"
try:
raw_posts = await scraper.scrape(platform, hashtags[0], num_posts)
return await _format_posts(raw_posts, platform)
except Exception as e:
print(f"Scraping failed: {e}")
return _fallback_data(num_posts)
def _format_posts(raw_posts, platform):
formatted = []
for post in raw_posts:
base_post = {
"caption": getattr(post, "rawContent", post.get('content', 'No caption')),
"image_url": getattr(post, "image_url", ""),
"video_url": "",
"audio_url": "",
"tagged_audience": [],
"date": str(time.strftime("%Y-%m-%d")),
"likes": getattr(post, "likeCount", 0),
"comments": getattr(post, "replyCount", 0)
}
formatted.append(base_post)
return formatted
def _fallback_data(num_posts):
return [
{
"caption": "Sample post about environmental issues",
"image_url": "https://example.com/sample.jpg",
"date": "2023-10-01",
"likes": 100,
"comments": 20,
} for _ in range(num_posts)
]
def extract_text_from_image(image_url):
try:
response = requests.get(image_url, timeout=10)
image = Image.open(BytesIO(response.content))
text = pytesseract.image_to_string(image)
return text.strip()
except Exception as e:
return f"OCR Error: {str(e)}"
def categorize_post(caption):
categories = ["activism", "politics", "social issues", "technology", "environment", "health"]
result = zero_shot_classifier(caption, candidate_labels=categories)
return result["labels"][0]
def analyze_sentiment(caption):
emotions = emotion_classifier(caption, top_k=None)
return sorted(emotions, key=lambda x: x["score"], reverse=True)[:3]
def detect_language(caption):
try:
return detect(caption)
except:
return "Unknown"
def extract_hashtags(caption):
return re.findall(r"#\w+", caption)
def process_posts(profile_url, hashtags, num_posts):
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
posts = loop.run_until_complete(extract_posts(profile_url, [h.strip() for h in hashtags.split(",")], num_posts))
doc = docx.Document()
doc.add_heading("Social Media Analysis Report", 0)
for i, post in enumerate(posts):
doc.add_heading(f"Post {i+1}", level=1)
# Metadata Section
meta = [
f"Date: {post.get('date', 'N/A')}",
f"Likes: {post.get('likes', 0)}",
f"Comments: {post.get('comments', 0)}",
f"Media: Pictures={1 if post['image_url'] else 0}, Videos={1 if post['video_url'] else 0}"
]
doc.add_paragraph("\n".join(meta))
# Content Analysis
content = doc.add_paragraph()
content.add_run("Caption Analysis:\n").bold = True
content.add_run(f"{post['caption']}\n\n")
# Sentiment and Language
content.add_run(f"Language: {detect_language(post['caption'])}\n")
emotions = analyze_sentiment(post['caption'])
content.add_run(f"Sentiment: {', '.join([f\"{e['label']} ({e['score']:.2f})\" for e in emotions])}\n")
# Hashtags and Category
hashtags = extract_hashtags(post['caption'])
content.add_run(f"Hashtags: {', '.join(hashtags) if hashtags else 'None'}\n")
content.add_run(f"Category: {categorize_post(post['caption'])}\n")
# Image Analysis
if post['image_url']:
img_analysis = doc.add_paragraph()
img_analysis.add_run("Image Analysis:\n").bold = True
img_analysis.add_run(f"Extracted Text: {extract_text_from_image(post['image_url'])[:500]}\n")
doc.add_page_break()
report_path = "social_media_analysis.docx"
doc.save(report_path)
return report_path
iface = gr.Interface(
fn=process_posts,
inputs=[
gr.Textbox(label="Profile URL", placeholder="Enter social media profile URL"),
gr.Textbox(label="Hashtags", placeholder="Comma-separated hashtags"),
gr.Slider(1, 50, value=5, label="Posts to Analyze")
],
outputs=gr.File(label="Download Report"),
title="Social Media Intelligence Analyzer",
description="""Real-time social media analysis with:
- 🕵️‍♂️ Live scraping
- 📊 Sentiment analysis
- 🖼️ Image OCR
- 🏷️ Hashtag tracking""",
examples=[
["https://twitter.com/eco_news", "climate, environment", 3],
["https://instagram.com/tech_innovators", "technology, future", 2]
]
)
if __name__ == "__main__":
iface.launch()