Spaces:
Runtime error
Runtime error
| 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() |