app.py

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()