app.py

import gradio as gr
import requests
import json
import pandas as pd
from datetime import datetime, timedelta
import plotly.express as px
import plotly.graph_objects as go
from typing import Dict, List, Any, Optional
import os
from dotenv import load_dotenv

# Load environment variables from .env file
load_dotenv()

# Configuration
PLAUSIBLE_URL = os.getenv("PLAUSIBLE_URL", "https://plausible.io/api/v2/query")
PLAUSIBLE_KEY = os.getenv("PLAUSIBLE_KEY")

class PlausibleAPI:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.headers = {
            'Authorization': f'Bearer {api_key}',
            'Content-Type': 'application/json'
        }
    
    def query(self, payload: Dict[str, Any]) -> Dict[str, Any]:
        """Make a query to the Plausible API"""

        if not self.api_key:
            return {"error": "PLAUSIBLE_KEY environment variable is not set"}
        
        try:
            response = requests.post(PLAUSIBLE_URL, headers=self.headers, json=payload)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            return {"error": f"API request failed: {str(e)}"}
        except json.JSONDecodeError as e:
            return {"error": f"Failed to parse JSON response: {str(e)}"}

# Initialize API client
api_client = PlausibleAPI(PLAUSIBLE_KEY)

def basic_stats_query(site_id: str, date_range: str, metrics: List[str]) -> tuple:
    """Get basic site statistics"""
    if not site_id:
        return "Please enter a site ID", None, None
    
    payload = {
        "site_id": site_id,
        "metrics": metrics,
        "date_range": date_range
    }
    
    result = api_client.query(payload)
    
    if "error" in result:
        return result["error"], None, None
    
    # Format results
    if result.get("results"):
        metrics_data = result["results"][0]["metrics"]
        stats_dict = dict(zip(metrics, metrics_data))
        
        # Create a simple bar chart
        fig = px.bar(
            x=list(stats_dict.keys()),
            y=list(stats_dict.values()),
            title=f"Stats for {site_id} ({date_range})"
        )
        fig.update_layout(xaxis_title="Metrics", yaxis_title="Values")
        
        return json.dumps(result, indent=2), stats_dict, fig
    
    return json.dumps(result, indent=2), None, None

def timeseries_query(site_id: str, date_range: str, metrics: List[str], time_dimension: str) -> tuple:
    """Get timeseries data"""
    if not site_id:
        return "Please enter a site ID", None
    
    payload = {
        "site_id": site_id,
        "metrics": metrics,
        "date_range": date_range,
        "dimensions": [time_dimension]
    }
    
    result = api_client.query(payload)
    
    if "error" in result:
        return result["error"], None
    
    # Create timeseries chart
    if result.get("results"):
        df_data = []
        for row in result["results"]:
            row_dict = {"time": row["dimensions"][0]}
            for i, metric in enumerate(metrics):
                row_dict[metric] = row["metrics"][i]
            df_data.append(row_dict)
        
        df = pd.DataFrame(df_data)
        df['time'] = pd.to_datetime(df['time'])
        
        fig = go.Figure()
        for metric in metrics:
            fig.add_trace(go.Scatter(
                x=df['time'],
                y=df[metric],
                mode='lines+markers',
                name=metric
            ))
        
        fig.update_layout(
            title=f"Timeseries for {site_id}",
            xaxis_title="Time",
            yaxis_title="Values"
        )
        
        return json.dumps(result, indent=2), fig
    
    return json.dumps(result, indent=2), None

def geographic_analysis(site_id: str, date_range: str, metrics: List[str]) -> tuple:
    """Analyze traffic by country and city"""
    if not site_id:
        return "Please enter a site ID", None, None
    
    payload = {
        "site_id": site_id,
        "metrics": metrics,
        "date_range": date_range,
        "dimensions": ["visit:country_name", "visit:city_name"],
        "filters": [["is_not", "visit:country_name", [""]]],
        "order_by": [[metrics[0], "desc"]]
    }
    
    result = api_client.query(payload)
    
    if "error" in result:
        return result["error"], None, None
    
    # Create geographic visualization
    if result.get("results"):
        df_data = []
        for row in result["results"]:
            row_dict = {
                "country": row["dimensions"][0],
                "city": row["dimensions"][1]
            }
            for i, metric in enumerate(metrics):
                row_dict[metric] = row["metrics"][i]
            df_data.append(row_dict)
        
        df = pd.DataFrame(df_data)
        
        # Create a bar chart of top countries
        country_stats = df.groupby('country')[metrics[0]].sum().sort_values(ascending=False).head(10)
        
        fig = px.bar(
            x=country_stats.index,
            y=country_stats.values,
            title=f"Top Countries by {metrics[0]} for {site_id}",
            labels={'x': 'Country', 'y': metrics[0]}
        )
        fig.update_xaxes(tickangle=45)
        
        return json.dumps(result, indent=2), fig, df.head(20).to_dict('records')
    
    return json.dumps(result, indent=2), None, None

def utm_analysis(site_id: str, date_range: str) -> tuple:
    """Analyze UTM parameters"""
    if not site_id:
        return "Please enter a site ID", None, None
    
    payload = {
        "site_id": site_id,
        "metrics": ["visitors", "events", "pageviews"],
        "date_range": date_range,
        "dimensions": ["visit:utm_medium", "visit:utm_source"],
        "filters": [["is_not", "visit:utm_medium", [""]]]
    }
    
    result = api_client.query(payload)
    
    if "error" in result:
        return result["error"], None, None
    
    if result.get("results"):
        df_data = []
        for row in result["results"]:
            df_data.append({
                "utm_medium": row["dimensions"][0] or "Direct",
                "utm_source": row["dimensions"][1] or "Direct",
                "visitors": row["metrics"][0],
                "events": row["metrics"][1],
                "pageviews": row["metrics"][2]
            })
        
        df = pd.DataFrame(df_data)
        
        # Create sunburst chart
        fig = px.sunburst(
            df,
            path=['utm_medium', 'utm_source'],
            values='visitors',
            title=f"UTM Analysis for {site_id}"
        )
        
        return json.dumps(result, indent=2), fig, df.to_dict('records')
    
    return json.dumps(result, indent=2), None, None

def custom_query(site_id: str, query_json: str) -> str:
    """Execute a custom JSON query"""
    if not site_id:
        return "Please enter a site ID"
    
    try:
        payload = json.loads(query_json)
        payload["site_id"] = site_id  # Override site_id
        
        result = api_client.query(payload)
        return json.dumps(result, indent=2)
    
    except json.JSONDecodeError as e:
        return f"Invalid JSON: {str(e)}"
    except Exception as e:
        return f"Error: {str(e)}"

# Gradio Interface
with gr.Blocks(title="Plausible Analytics Dashboard", theme=gr.themes.Soft()) as demo:
    gr.Markdown("# 📊 Plausible Analytics Dashboard")
    gr.Markdown("MCP Server to analyze your website statistics using the Plausible Stats API.\n\nSo far this app is 100% vibe coded with the help of Claude Sonnet 4.\n\nTry it out with the site id 'azettl.net' or 'fridgeleftoversai.com'.")
    
    with gr.Tab("Basic Stats"):
        gr.Markdown("### Get basic website statistics")
        
        with gr.Row():
            site_input = gr.Textbox(
                label="Site ID",
                placeholder="example.com",
                info="Your domain as added to Plausible"
            )
            date_range = gr.Dropdown(
                choices=["day", "7d", "28d", "30d", "month", "6mo", "12mo", "year", "all"],
                value="7d",
                label="Date Range"
            )
        
        metrics_input = gr.CheckboxGroup(
            choices=["visitors", "visits", "pageviews", "views_per_visit", "bounce_rate", "visit_duration", "events"],
            value=["visitors", "pageviews", "bounce_rate"],
            label="Metrics to Analyze"
        )
        
        basic_btn = gr.Button("Get Basic Stats", variant="primary")
        
        with gr.Row():
            basic_json = gr.Code(label="API Response", language="json")
            basic_stats = gr.JSON(label="Stats Summary")
        
        basic_chart = gr.Plot(label="Statistics Chart")
        
        basic_btn.click(
            basic_stats_query,
            inputs=[site_input, date_range, metrics_input],
            outputs=[basic_json, basic_stats, basic_chart]
        )
    
    with gr.Tab("Timeseries"):
        gr.Markdown("### View trends over time")
        
        with gr.Row():
            ts_site = gr.Textbox(label="Site ID", placeholder="example.com")
            ts_date_range = gr.Dropdown(
                choices=["day", "7d", "28d", "30d", "month"],
                value="7d",
                label="Date Range"
            )
        
        with gr.Row():
            ts_metrics = gr.CheckboxGroup(
                choices=["visitors", "visits", "pageviews", "events"],
                value=["visitors", "pageviews"],
                label="Metrics"
            )
            ts_time_dim = gr.Dropdown(
                choices=["time:hour", "time:day", "time:week", "time:month"],
                value="time:day",
                label="Time Dimension"
            )
        
        ts_btn = gr.Button("Generate Timeseries", variant="primary")
        
        with gr.Row():
            ts_json = gr.Code(label="API Response", language="json")
            ts_chart = gr.Plot(label="Timeseries Chart")
        
        ts_btn.click(
            timeseries_query,
            inputs=[ts_site, ts_date_range, ts_metrics, ts_time_dim],
            outputs=[ts_json, ts_chart]
        )
    
    with gr.Tab("Geographic Analysis"):
        gr.Markdown("### Analyze traffic by location")
        
        with gr.Row():
            geo_site = gr.Textbox(label="Site ID", placeholder="example.com")
            geo_date_range = gr.Dropdown(
                choices=["day", "7d", "28d", "30d", "month"],
                value="7d",
                label="Date Range"
            )
        
        geo_metrics = gr.CheckboxGroup(
            choices=["visitors", "visits", "pageviews", "bounce_rate"],
            value=["visitors", "pageviews"],
            label="Metrics"
        )
        
        geo_btn = gr.Button("Analyze Geography", variant="primary")
        
        with gr.Row():
            geo_json = gr.Code(label="API Response", language="json")
            geo_chart = gr.Plot(label="Geographic Chart")
        
        geo_table = gr.DataFrame(label="Top Locations")
        
        geo_btn.click(
            geographic_analysis,
            inputs=[geo_site, geo_date_range, geo_metrics],
            outputs=[geo_json, geo_chart, geo_table]
        )
    
    with gr.Tab("UTM Analysis"):
        gr.Markdown("### Analyze marketing campaigns and traffic sources")
        
        with gr.Row():
            utm_site = gr.Textbox(label="Site ID", placeholder="example.com")
            utm_date_range = gr.Dropdown(
                choices=["day", "7d", "28d", "30d", "month"],
                value="7d",
                label="Date Range"
            )
        
        utm_btn = gr.Button("Analyze UTM Parameters", variant="primary")
        
        with gr.Row():
            utm_json = gr.Code(label="API Response", language="json")
            utm_chart = gr.Plot(label="UTM Sunburst Chart")
        
        utm_table = gr.DataFrame(label="UTM Data")
        
        utm_btn.click(
            utm_analysis,
            inputs=[utm_site, utm_date_range],
            outputs=[utm_json, utm_chart, utm_table]
        )
    
    with gr.Tab("Custom Query"):
        gr.Markdown("### Execute custom JSON queries")
        gr.Markdown("Use this tab to run advanced queries with custom filters and dimensions.")
        
        custom_site = gr.Textbox(label="Site ID", placeholder="example.com")
        
        custom_query_input = gr.Code(
            label="JSON Query",
            language="json",
            value="""{
  "metrics": ["visitors", "pageviews"],
  "date_range": "7d",
  "dimensions": ["visit:source"],
  "order_by": [["visitors", "desc"]],
  "pagination": {"limit": 10}
}""",
            lines=15
        )
        
        custom_btn = gr.Button("Execute Query", variant="primary")
        custom_result = gr.Code(label="Query Result", language="json", lines=20)
        
        custom_btn.click(
            custom_query,
            inputs=[custom_site, custom_query_input],
            outputs=[custom_result]
        )
    
    with gr.Tab("Setup & Documentation"):
        gr.Markdown("""
        ## 🔧 Setup Instructions
        
        ### For Personal Use (Recommended)
        This MCP server is designed for **personal use only**. Each user should run their own instance.
        
        **Setup Steps:**
        1. **Get your Plausible API key:**
           - Log into your Plausible account
           - Go to Account Settings → API Keys
           - Create a new key, select "Stats API" as type
        
        2. **Set environment variable:**
           ```bash
           # Windows
           set PLAUSIBLE_KEY=your-key-here
           
           # Mac/Linux
           export PLAUSIBLE_KEY=your-key-here
           
           # Or create .env file:
           echo "PLAUSIBLE_KEY=your-key-here" > .env
           ```
        
        2. **Install dependencies:**
           ```bash    
           pip install -r requirements.txt
           ```

        3. **Run the server:**
           ```bash
           python app.py
           ```
        
        4. **Add to Claude Desktop config:**
           ```json
           {
             "mcpServers": {
               "plausible": {
                 "command": "npx",
                 "args": ["mcp-remote", "http://localhost:7860/gradio_api/mcp/sse"]
               }
             }
           }
           ```
        
        ### ⚠️ Security Notice
        - **DO NOT** share your API key with others
        - **DO NOT** run this as a public server with your API key (Like I do here to show you how it works 🙈)
        - Each user should run their own instance with their own API key
        
        ---
        
        ## 📖 API Reference
        
        **Available Metrics:**
        - `visitors`: Unique visitors
        - `visits`: Number of sessions
        - `pageviews`: Total page views
        - `views_per_visit`: Average pages per session
        - `bounce_rate`: Bounce rate percentage
        - `visit_duration`: Average visit duration
        - `events`: Total events
        
        **Date Ranges:**
        - `day`: Current day
        - `7d`: Last 7 days
        - `28d`: Last 28 days
        - `30d`: Last 30 days
        - `month`: Current month
        - `6mo`: Last 6 months
        - `12mo`: Last 12 months
        - `year`: Current year
        - `all`: All time
        
        **Common Dimensions:**
        - `visit:country_name`: Country
        - `visit:source`: Traffic source
        - `visit:device`: Device type
        - `visit:browser`: Browser
        - `event:page`: Page path
        - `time:day`: Daily grouping
        - `time:hour`: Hourly grouping
        
        **Example Filters:**
        ```json
        [["is", "visit:country_name", ["United States", "Canada"]]]
        [["contains", "event:page", ["/blog"]]]
        [["is_not", "visit:device", ["Mobile"]]]
        ```
        """)

# Launch configuration
if __name__ == "__main__":
    demo.launch(
        server_name="0.0.0.0",
        server_port=7860,
        share=False,
        debug=False,
        mcp_server=True
    )