Google Analytics

# Google Analytics API Configuration
# Copy this file to .env and fill in your actual values

# Your Google Analytics 4 Property ID
# Find this in GA4: Admin > Property Settings > Property ID
# Format: 123456789 (numeric only, not the "G-" measurement ID)
GOOGLE_ANALYTICS_PROPERTY_ID=your-property-id-here

# Path to your Google Cloud service account JSON key file
# Create a service account in Google Cloud Console and download the JSON key
# Store this file OUTSIDE your git repository for security
# Example: /home/user/.config/google-analytics/service-account-key.json
GOOGLE_APPLICATION_CREDENTIALS=/path/to/service-account-key.json

# How to set up:
# 1. Go to Google Cloud Console (console.cloud.google.com)
# 2. Create or select a project
# 3. Enable the Google Analytics Data API
# 4. Create a service account (IAM & Admin > Service Accounts)
# 5. Download the JSON key file
# 6. Go to GA4 Admin > Property Access Management
# 7. Add the service account email with "Viewer" role
# 8. Update the values above with your property ID and key file path
# 9. Rename this file to .env
# 10. NEVER commit .env or the JSON key file to version control!

# Google Analytics Skill Dependencies
# Install with: pip install -r requirements.txt

google-analytics-data>=0.17.0
python-dotenv>=1.0.0
pandas>=2.0.0

#!/usr/bin/env python3
"""
Google Analytics Data Analysis Tool

Performs higher-level analysis on Google Analytics data including:
- Period comparisons (current vs previous)
- Trend detection
- Performance insights
- Automated recommendations

Usage:
    python analyze.py --period last-30-days --compare previous-period
    python analyze.py --analysis-type traffic-sources --days 30
    python analyze.py --analysis-type funnel --steps "homepage,/products,/cart,/checkout"
"""

import os
import sys
import json
import argparse
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Tuple

try:
    from ga_client import GoogleAnalyticsClient
except ImportError:
    print("Error: ga_client.py not found in the same directory", file=sys.stderr)
    sys.exit(1)


class AnalyticsAnalyzer:
    """Performs analysis on Google Analytics data."""

    def __init__(self):
        """Initialize the analyzer with GA client."""
        self.client = GoogleAnalyticsClient()

    def compare_periods(
        self, current_days: int = 30, metrics: Optional[List[str]] = None
    ) -> Dict:
        """
        Compare current period with previous period.

        Args:
            current_days: Number of days in current period
            metrics: List of metrics to compare (default: core metrics)

        Returns:
            Dictionary with comparison data and insights
        """
        if metrics is None:
            metrics = [
                "sessions",
                "activeUsers",
                "newUsers",
                "bounceRate",
                "engagementRate",
                "averageSessionDuration",
            ]

        # Fetch current period
        current = self.client.run_report(
            start_date=f"{current_days}daysAgo",
            end_date="yesterday",
            metrics=metrics,
            limit=1,
        )

        # Fetch previous period
        previous_start = current_days * 2
        previous_end = current_days + 1
        previous = self.client.run_report(
            start_date=f"{previous_start}daysAgo",
            end_date=f"{previous_end}daysAgo",
            metrics=metrics,
            limit=1,
        )

        # Calculate changes
        comparison = {
            "current_period": f"Last {current_days} days",
            "previous_period": f"Previous {current_days} days",
            "metrics": {},
        }

        if current["totals"] and previous["totals"]:
            for i, metric in enumerate(metrics):
                current_val = float(current["totals"][i]["value"])
                previous_val = float(previous["totals"][i]["value"])

                # Calculate percentage change
                if previous_val != 0:
                    change_pct = ((current_val - previous_val) / previous_val) * 100
                else:
                    change_pct = 0

                comparison["metrics"][metric] = {
                    "current": current_val,
                    "previous": previous_val,
                    "change": current_val - previous_val,
                    "change_percent": round(change_pct, 2),
                }

        # Generate insights
        comparison["insights"] = self._generate_insights(comparison["metrics"])

        return comparison

    def analyze_traffic_sources(self, days: int = 30, limit: int = 20) -> Dict:
        """
        Analyze traffic sources and their performance.

        Args:
            days: Number of days to analyze
            limit: Number of sources to return

        Returns:
            Dictionary with source performance data and recommendations
        """
        result = self.client.run_report(
            start_date=f"{days}daysAgo",
            end_date="yesterday",
            metrics=["sessions", "engagementRate", "bounceRate", "conversions"],
            dimensions=["sessionSource", "sessionMedium"],
            limit=limit,
            order_by="-sessions",
        )

        # Analyze sources
        sources = []
        for row in result["rows"]:
            source = row["dimensions"]["sessionSource"]
            medium = row["dimensions"]["sessionMedium"]
            sessions = int(row["metrics"]["sessions"])
            engagement = float(row["metrics"]["engagementRate"])
            bounce = float(row["metrics"]["bounceRate"])
            conversions = int(row["metrics"].get("conversions", 0))

            conv_rate = (conversions / sessions * 100) if sessions > 0 else 0

            sources.append(
                {
                    "source": source,
                    "medium": medium,
                    "sessions": sessions,
                    "engagement_rate": round(engagement * 100, 2),
                    "bounce_rate": round(bounce * 100, 2),
                    "conversions": conversions,
                    "conversion_rate": round(conv_rate, 2),
                }
            )

        analysis = {
            "period": f"Last {days} days",
            "sources": sources,
            "recommendations": self._recommend_source_optimizations(sources),
        }

        return analysis

    def analyze_content_performance(self, days: int = 30, limit: int = 50) -> Dict:
        """
        Analyze page performance and identify issues.

        Args:
            days: Number of days to analyze
            limit: Number of pages to return

        Returns:
            Dictionary with page performance and improvement opportunities
        """
        result = self.client.run_report(
            start_date=f"{days}daysAgo",
            end_date="yesterday",
            metrics=[
                "screenPageViews",
                "bounceRate",
                "averageSessionDuration",
                "conversions",
            ],
            dimensions=["pagePath", "pageTitle"],
            limit=limit,
            order_by="-screenPageViews",
        )

        # Identify high-bounce pages
        high_bounce_threshold = 0.6
        problem_pages = []

        for row in result["rows"]:
            page_path = row["dimensions"]["pagePath"]
            page_title = row["dimensions"]["pageTitle"]
            views = int(row["metrics"]["screenPageViews"])
            bounce = float(row["metrics"]["bounceRate"])
            avg_duration = float(row["metrics"]["averageSessionDuration"])

            if bounce > high_bounce_threshold and views > 100:
                problem_pages.append(
                    {
                        "path": page_path,
                        "title": page_title,
                        "views": views,
                        "bounce_rate": round(bounce * 100, 2),
                        "avg_duration": round(avg_duration, 2),
                        "issue": self._diagnose_page_issue(bounce, avg_duration),
                    }
                )

        analysis = {
            "period": f"Last {days} days",
            "total_pages": result["row_count"],
            "high_bounce_pages": len(problem_pages),
            "problem_pages": problem_pages[:10],  # Top 10 issues
            "recommendations": self._recommend_content_improvements(problem_pages),
        }

        return analysis

    def analyze_device_performance(self, days: int = 30) -> Dict:
        """
        Compare performance across device types.

        Args:
            days: Number of days to analyze

        Returns:
            Dictionary with device performance comparison
        """
        result = self.client.run_report(
            start_date=f"{days}daysAgo",
            end_date="yesterday",
            metrics=[
                "sessions",
                "bounceRate",
                "averageSessionDuration",
                "conversions",
                "engagementRate",
            ],
            dimensions=["deviceCategory"],
            limit=10,
            order_by="-sessions",
        )

        devices = []
        for row in result["rows"]:
            device = row["dimensions"]["deviceCategory"]
            sessions = int(row["metrics"]["sessions"])
            bounce = float(row["metrics"]["bounceRate"])
            duration = float(row["metrics"]["averageSessionDuration"])
            conversions = int(row["metrics"].get("conversions", 0))
            engagement = float(row["metrics"]["engagementRate"])

            conv_rate = (conversions / sessions * 100) if sessions > 0 else 0

            devices.append(
                {
                    "device": device,
                    "sessions": sessions,
                    "bounce_rate": round(bounce * 100, 2),
                    "avg_duration": round(duration, 2),
                    "conversion_rate": round(conv_rate, 2),
                    "engagement_rate": round(engagement * 100, 2),
                }
            )

        analysis = {
            "period": f"Last {days} days",
            "devices": devices,
            "recommendations": self._recommend_device_optimizations(devices),
        }

        return analysis

    def _generate_insights(self, metrics: Dict) -> List[str]:
        """Generate insights from metric comparisons."""
        insights = []

        for metric, data in metrics.items():
            change_pct = data["change_percent"]

            if abs(change_pct) < 2:
                status = "stable"
            elif change_pct > 0:
                status = "improving"
            else:
                status = "declining"

            # Add insights for significant changes
            if abs(change_pct) >= 5:
                direction = "increased" if change_pct > 0 else "decreased"
                insights.append(
                    f"{metric.replace('_', ' ').title()}: {direction} by {abs(change_pct):.1f}%"
                )

        return insights

    def _recommend_source_optimizations(self, sources: List[Dict]) -> List[Dict]:
        """Generate recommendations for traffic source optimization."""
        recommendations = []

        if not sources:
            return recommendations

        # Find best performing source
        best_source = max(sources, key=lambda x: x["conversion_rate"])
        recommendations.append(
            {
                "priority": "HIGH",
                "action": f"Scale {best_source['source']}/{best_source['medium']}",
                "reason": f"Highest conversion rate ({best_source['conversion_rate']}%)",
                "expected_impact": "Increase overall conversions by 20-30%",
            }
        )

        # Find high-traffic low-conversion sources
        for source in sources[:5]:  # Check top 5
            if source["conversion_rate"] < 2.0 and source["sessions"] > 1000:
                recommendations.append(
                    {
                        "priority": "MEDIUM",
                        "action": f"Optimize {source['source']}/{source['medium']}",
                        "reason": f"High traffic ({source['sessions']} sessions) but low conversion ({source['conversion_rate']}%)",
                        "expected_impact": "Potential conversion rate improvement of 50-100%",
                    }
                )

        return recommendations

    def _recommend_content_improvements(self, problem_pages: List[Dict]) -> List[Dict]:
        """Generate recommendations for content improvements."""
        recommendations = []

        if not problem_pages:
            recommendations.append(
                {
                    "priority": "INFO",
                    "action": "Content performing well",
                    "reason": "No pages with critically high bounce rates",
                    "expected_impact": "Continue monitoring",
                }
            )
            return recommendations

        # Prioritize by traffic
        problem_pages.sort(key=lambda x: x["views"], reverse=True)

        for page in problem_pages[:3]:  # Top 3 issues
            recommendations.append(
                {
                    "priority": "HIGH",
                    "action": f"Improve {page['path']}",
                    "reason": f"{page['issue']} ({page['bounce_rate']}% bounce rate)",
                    "expected_impact": "Reduce bounce rate by 20-30%",
                }
            )

        return recommendations

    def _recommend_device_optimizations(self, devices: List[Dict]) -> List[Dict]:
        """Generate recommendations for device optimization."""
        recommendations = []

        if len(devices) < 2:
            return recommendations

        # Compare mobile vs desktop
        mobile = next((d for d in devices if d["device"] == "mobile"), None)
        desktop = next((d for d in devices if d["device"] == "desktop"), None)

        if mobile and desktop:
            conv_diff = (
                (desktop["conversion_rate"] - mobile["conversion_rate"])
                / desktop["conversion_rate"]
                * 100
            )

            if conv_diff > 30:  # Desktop significantly better
                recommendations.append(
                    {
                        "priority": "CRITICAL",
                        "action": "Mobile experience optimization",
                        "reason": f"Mobile conversion rate {mobile['conversion_rate']}% vs desktop {desktop['conversion_rate']}%",
                        "expected_impact": "Improve mobile conversion by 30-50%",
                    }
                )

        return recommendations

    def _diagnose_page_issue(self, bounce_rate: float, avg_duration: float) -> str:
        """Diagnose the issue with a high-bounce page."""
        if bounce_rate > 0.7 and avg_duration < 30:
            return "Content mismatch - users leave quickly"
        elif bounce_rate > 0.6 and avg_duration > 60:
            return "Missing CTA - users read but don't act"
        elif bounce_rate > 0.6:
            return "High bounce - needs investigation"
        else:
            return "Performance issue"


def main():
    parser = argparse.ArgumentParser(
        description="Analyze Google Analytics data",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )

    parser.add_argument(
        "--analysis-type",
        choices=["overview", "sources", "content", "devices"],
        default="overview",
        help="Type of analysis to perform",
    )
    parser.add_argument(
        "--days", type=int, default=30, help="Number of days to analyze (default: 30)"
    )
    parser.add_argument(
        "--compare",
        action="store_true",
        help="Compare with previous period",
    )
    parser.add_argument(
        "--output", help="Output file path (default: stdout)"
    )

    args = parser.parse_args()

    try:
        analyzer = AnalyticsAnalyzer()

        # Run analysis
        if args.analysis_type == "overview" and args.compare:
            result = analyzer.compare_periods(current_days=args.days)
        elif args.analysis_type == "sources":
            result = analyzer.analyze_traffic_sources(days=args.days)
        elif args.analysis_type == "content":
            result = analyzer.analyze_content_performance(days=args.days)
        elif args.analysis_type == "devices":
            result = analyzer.analyze_device_performance(days=args.days)
        else:
            result = analyzer.compare_periods(current_days=args.days)

        # Format output
        output = json.dumps(result, indent=2)

        # Write output
        if args.output:
            with open(args.output, "w") as f:
                f.write(output)
            print(f"Analysis saved to {args.output}", file=sys.stderr)
        else:
            print(output)

    except Exception as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Google Analytics 4 Data API Client

Fetches analytics data from Google Analytics 4 using the Data API.
Authentication uses service account credentials from environment variables.

Usage:
    python ga_client.py --days 30 --metrics sessions,users
    python ga_client.py --start 2026-01-01 --end 2026-01-31 --dimensions country
    python ga_client.py --days 7 --metrics sessions --dimensions pagePath --limit 10

Environment Variables:
    GOOGLE_ANALYTICS_PROPERTY_ID: GA4 property ID (required)
    GOOGLE_APPLICATION_CREDENTIALS: Path to service account JSON (required)
"""

import os
import sys
import json
import argparse
from datetime import datetime, timedelta
from typing import List, Dict, Optional

try:
    from google.analytics.data_v1beta import BetaAnalyticsDataClient
    from google.analytics.data_v1beta.types import (
        DateRange,
        Dimension,
        Metric,
        RunReportRequest,
        OrderBy,
        FilterExpression,
        Filter,
    )
    from dotenv import load_dotenv
except ImportError as e:
    print(f"Error: Required package not installed: {e}", file=sys.stderr)
    print("Install with: pip install google-analytics-data python-dotenv", file=sys.stderr)
    sys.exit(1)


class GoogleAnalyticsClient:
    """Client for interacting with Google Analytics 4 Data API."""

    def __init__(self):
        """Initialize the client with credentials from environment."""
        load_dotenv()  # Load from .env file if present

        self.property_id = os.environ.get("GOOGLE_ANALYTICS_PROPERTY_ID")
        if not self.property_id:
            raise ValueError(
                "GOOGLE_ANALYTICS_PROPERTY_ID environment variable not set. "
                "Find your property ID in GA4: Admin > Property Settings"
            )

        credentials_path = os.environ.get("GOOGLE_APPLICATION_CREDENTIALS")
        if not credentials_path:
            raise ValueError(
                "GOOGLE_APPLICATION_CREDENTIALS environment variable not set. "
                "Set it to the path of your service account JSON file."
            )

        if not os.path.exists(credentials_path):
            raise FileNotFoundError(
                f"Service account file not found: {credentials_path}"
            )

        try:
            self.client = BetaAnalyticsDataClient()
        except Exception as e:
            raise RuntimeError(
                f"Failed to initialize Google Analytics client: {e}\n"
                "Verify your service account has access to the GA4 property."
            )

    def run_report(
        self,
        start_date: str,
        end_date: str,
        metrics: List[str],
        dimensions: Optional[List[str]] = None,
        limit: int = 10,
        order_by: Optional[str] = None,
        filter_expression: Optional[str] = None,
    ) -> Dict:
        """
        Run a report query against Google Analytics.

        Args:
            start_date: Start date (YYYY-MM-DD or 'NdaysAgo')
            end_date: End date (YYYY-MM-DD or 'today'/'yesterday')
            metrics: List of metric names (e.g., ['sessions', 'users'])
            dimensions: List of dimension names (e.g., ['country', 'city'])
            limit: Maximum number of rows to return
            order_by: Metric or dimension to sort by
            filter_expression: Filter to apply (dimension_name:value)

        Returns:
            Dictionary with report data and metadata
        """
        # Build request
        request = RunReportRequest(
            property=f"properties/{self.property_id}",
            date_ranges=[DateRange(start_date=start_date, end_date=end_date)],
            metrics=[Metric(name=m) for m in metrics],
            dimensions=[Dimension(name=d) for d in (dimensions or [])],
            limit=limit,
        )

        # Add ordering
        if order_by:
            desc = True
            if order_by.startswith("+"):
                desc = False
                order_by = order_by[1:]
            elif order_by.startswith("-"):
                order_by = order_by[1:]

            # Check if it's a metric or dimension
            if order_by in metrics:
                request.order_bys = [
                    OrderBy(metric=OrderBy.MetricOrderBy(metric_name=order_by), desc=desc)
                ]
            elif dimensions and order_by in dimensions:
                request.order_bys = [
                    OrderBy(
                        dimension=OrderBy.DimensionOrderBy(dimension_name=order_by),
                        desc=desc,
                    )
                ]

        # Add filter
        if filter_expression and ":" in filter_expression:
            field_name, value = filter_expression.split(":", 1)
            request.dimension_filter = FilterExpression(
                filter=Filter(
                    field_name=field_name,
                    string_filter=Filter.StringFilter(value=value),
                )
            )

        try:
            response = self.client.run_report(request)
        except Exception as e:
            raise RuntimeError(f"Failed to run report: {e}")

        # Parse response
        return self._parse_response(response)

    def _parse_response(self, response) -> Dict:
        """Parse API response into a structured dictionary."""
        result = {
            "dimension_headers": [h.name for h in response.dimension_headers],
            "metric_headers": [
                {"name": h.name, "type": h.type_.name} for h in response.metric_headers
            ],
            "rows": [],
            "row_count": response.row_count,
            "metadata": {},
        }

        # Add totals if present
        if response.totals:
            result["totals"] = [
                {"value": v.value} for v in response.totals[0].metric_values
            ]

        # Parse rows
        for row in response.rows:
            parsed_row = {
                "dimensions": {},
                "metrics": {},
            }

            # Dimension values
            for i, value in enumerate(row.dimension_values):
                dim_name = result["dimension_headers"][i]
                parsed_row["dimensions"][dim_name] = value.value

            # Metric values
            for i, value in enumerate(row.metric_values):
                metric_info = result["metric_headers"][i]
                metric_name = metric_info["name"]
                parsed_row["metrics"][metric_name] = value.value

            result["rows"].append(parsed_row)

        return result


def parse_date_range(days: Optional[int], start: Optional[str], end: Optional[str]):
    """Parse date range arguments into start and end dates."""
    if days:
        return f"{days}daysAgo", "yesterday"
    elif start and end:
        return start, end
    else:
        # Default to last 7 days
        return "7daysAgo", "yesterday"


def main():
    parser = argparse.ArgumentParser(
        description="Fetch Google Analytics 4 data",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  # Last 30 days of sessions and users
  python ga_client.py --days 30 --metrics sessions,users

  # Specific date range with dimensions
  python ga_client.py --start 2026-01-01 --end 2026-01-31 \\
      --metrics sessions,bounceRate --dimensions country,city

  # Top pages by views
  python ga_client.py --days 7 --metrics screenPageViews \\
      --dimensions pagePath --order-by screenPageViews --limit 20

  # Filter by country
  python ga_client.py --days 30 --metrics sessions \\
      --dimensions country --filter "country:United States"
        """,
    )

    # Date range arguments
    date_group = parser.add_mutually_exclusive_group()
    date_group.add_argument(
        "--days", type=int, help="Number of days to look back (e.g., 30)"
    )
    date_group.add_argument(
        "--start", help="Start date (YYYY-MM-DD or 'NdaysAgo')"
    )
    parser.add_argument("--end", help="End date (YYYY-MM-DD or 'today'/'yesterday')")

    # Query arguments
    parser.add_argument(
        "--metrics",
        required=True,
        help="Comma-separated list of metrics (e.g., sessions,users,bounceRate)",
    )
    parser.add_argument(
        "--dimensions",
        help="Comma-separated list of dimensions (e.g., country,city,deviceCategory)",
    )
    parser.add_argument(
        "--limit", type=int, default=10, help="Maximum rows to return (default: 10)"
    )
    parser.add_argument(
        "--order-by",
        help="Metric or dimension to sort by (prefix with - for desc, + for asc)",
    )
    parser.add_argument(
        "--filter", help="Filter expression (e.g., 'country:United States')"
    )

    # Output arguments
    parser.add_argument(
        "--format",
        choices=["json", "table"],
        default="json",
        help="Output format (default: json)",
    )
    parser.add_argument(
        "--output", help="Output file path (default: stdout)"
    )

    args = parser.parse_args()

    try:
        # Initialize client
        client = GoogleAnalyticsClient()

        # Parse date range
        start_date, end_date = parse_date_range(args.days, args.start, args.end)

        # Parse metrics and dimensions
        metrics = [m.strip() for m in args.metrics.split(",")]
        dimensions = (
            [d.strip() for d in args.dimensions.split(",")] if args.dimensions else None
        )

        # Run report
        result = client.run_report(
            start_date=start_date,
            end_date=end_date,
            metrics=metrics,
            dimensions=dimensions,
            limit=args.limit,
            order_by=args.order_by,
            filter_expression=args.filter,
        )

        # Format output
        if args.format == "json":
            output = json.dumps(result, indent=2)
        else:  # table format
            output = format_as_table(result)

        # Write output
        if args.output:
            with open(args.output, "w") as f:
                f.write(output)
            print(f"Report saved to {args.output}", file=sys.stderr)
        else:
            print(output)

    except Exception as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)


def format_as_table(result: Dict) -> str:
    """Format result as a human-readable table."""
    lines = []

    # Header
    headers = result["dimension_headers"] + [m["name"] for m in result["metric_headers"]]
    lines.append(" | ".join(headers))
    lines.append("-" * (len(" | ".join(headers))))

    # Rows
    for row in result["rows"]:
        values = []
        for dim in result["dimension_headers"]:
            values.append(row["dimensions"].get(dim, ""))
        for metric in result["metric_headers"]:
            values.append(row["metrics"].get(metric["name"], ""))
        lines.append(" | ".join(values))

    # Totals
    if "totals" in result:
        lines.append("-" * (len(" | ".join(headers))))
        total_values = ["TOTAL"] + [""] * (len(result["dimension_headers"]) - 1)
        total_values += [t["value"] for t in result["totals"]]
        lines.append(" | ".join(total_values))

    lines.append("")
    lines.append(f"Total rows: {result['row_count']}")

    return "\n".join(lines)


if __name__ == "__main__":
    main()