Building a Real-Time Social Listening & Brand Sentiment Engine with Python, Gemini 1.5, and DuckDB

import duckdb
import hashlib

class StorageManager:
    def __init__(self, db_path: str = "brand_intel.duckdb"):
        self.con = duckdb.connect(db_path)
        self._init_schema()

    def _init_schema(self):
        self.con.execute("""
            CREATE TABLE IF NOT EXISTS raw_mentions (
                id VARCHAR PRIMARY KEY,
                platform VARCHAR,
                author VARCHAR,
                content VARCHAR,
                url VARCHAR,
                created_at TIMESTAMP,
                processed BOOLEAN DEFAULT FALSE
            );
            CREATE TABLE IF NOT EXISTS sentiment_results (
                id VARCHAR PRIMARY KEY,
                polarity FLOAT,
                urgency VARCHAR,
                is_sarcastic BOOLEAN,
                entities JSON,
                summary VARCHAR,
                evaluated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                FOREIGN KEY (id) REFERENCES raw_mentions(id)
            );
        """)

    def is_duplicate(self, platform: str, external_id: str) -> bool:
        uid = hashlib.sha256(f"{platform}:{external_id}".encode()).hexdigest()
        res = self.con.execute("SELECT 1 FROM raw_mentions WHERE id = ?", [uid]).fetchone()
        return res is not None

    def insert_mention(self, platform: str, external_id: str, author: str, content: str, url: str, created_at):
        uid = hashlib.sha256(f"{platform}:{external_id}".encode()).hexdigest()
        self.con.execute("""
            INSERT INTO raw_mentions (id, platform, author, content, url, created_at)
            VALUES (?, ?, ?, ?, ?, ?)
            ON CONFLICT (id) DO NOTHING
        """, [uid, platform, author, content, url, created_at])
        return uid

from pydantic import BaseModel, Field
from typing import List, Literal
import google.generativeai as genai
import os

genai.configure(api_key=os.environ["GEMINI_API_KEY"])

class SentimentEvaluation(BaseModel):
    polarity: float = Field(
        description="Polarity score strictly between -1.0 (extremely negative/hostile) and 1.0 (enthusiastic/positive)."
    )
    urgency: Literal["low", "medium", "high", "critical"] = Field(
        description="'critical' indicates data breaches, service downtime, or legal threats; 'high' indicates severe bugs or churn threats; 'medium' is regular feedback; 'low' is passive mention."
    )
    is_sarcastic: bool = Field(description="True if the statement implies the opposite of its literal words.")
    entities: List[str] = Field(description="List of specific products, competitor names, or people mentioned.")
    summary: str = Field(description="1-sentence technical digest of the user's issue or praise.")

import json

def evaluate_mention(brand_name: str, text: str) -> SentimentEvaluation:
    prompt = f"""
    Analyze the following social mention referencing the brand: '{brand_name}'.
    Contextualize sarcasm, tech nuances, and developer sentiment.

    Content:
    """{text}"""
    """

    model = genai.GenerativeModel("gemini-1.5-flash")

    response = model.generate_content(
        prompt,
        generation_config=genai.GenerationConfig(
            response_mime_type="application/json",
            response_schema=SentimentEvaluation,
            temperature=0.1,
        ),
    )

    return SentimentEvaluation.model_validate_json(response.text)

import httpx
import asyncio

SLACK_WEBHOOK_URL = os.environ.get("SLACK_WEBHOOK_URL")

async def dispatch_alert(eval_data: SentimentEvaluation, platform: str, url: str, content: str):
    # Alert trigger: high/critical urgency OR heavily negative sentiment
    if eval_data.urgency in ["high", "critical"] or eval_data.polarity <= -0.5:
        color = "#FF0000" if eval_data.urgency == "critical" else "#FFA500"

        payload = {
            "attachments": [
                {
                    "color": color,
                    "title": f"🚨 Brand Alert [{eval_data.urgency.upper()}] on {platform}",
                    "title_link": url,
                    "fields": [
                        {"title": "Summary", "value": eval_data.summary, "short": False},
                        {"title": "Polarity", "value": f"{eval_data.polarity:.2f}", "short": True},
                        {"title": "Sarcasm Detected", "value": str(eval_data.is_sarcastic), "short": True},
                        {"title": "Entities", "value": ", ".join(eval_data.entities) or "None", "short": True},
                    ],
                    "text": f"*Snippet:*
> {content[:280]}...",
                }
            ]
        }

        async with httpx.AsyncClient() as client:
            resp = await client.post(SLACK_WEBHOOK_URL, json=payload)
            resp.raise_for_status()

async def analysis_worker(queue: asyncio.Queue, db: StorageManager, brand_name: str):
    while True:
        item = await queue.get()
        uid, platform, author, content, url = item
        try:
            # Non-blocking run for Gemini evaluation
            eval_res = await asyncio.to_thread(evaluate_mention, brand_name, content)

            # Save analytics to DuckDB
            db.con.execute("""
                INSERT INTO sentiment_results (id, polarity, urgency, is_sarcastic, entities, summary)
                VALUES (?, ?, ?, ?, ?, ?)
            """, [uid, eval_res.polarity, eval_res.urgency, eval_res.is_sarcastic, 
                  json.dumps(eval_res.entities), eval_res.summary])

            # Route Alerts
            await dispatch_alert(eval_res, platform, url, content)

        except Exception as e:
            print(f"[ERROR] Failed to process {uid}: {e}")
        finally:
            queue.task_done()

SELECT 
    date_trunc('hour', evaluated_at) as hour_bucket,
    AVG(polarity) as avg_sentiment,
    COUNT(CASE WHEN urgency IN ('high', 'critical') THEN 1 END) as critical_count,
    COUNT(*) as total_mentions
FROM sentiment_results
WHERE evaluated_at >= NOW() - INTERVAL '24 HOURS'
GROUP BY 1
ORDER BY 1 DESC;