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!

MIT License

Copyright (c) 2026 Rebecca Rae Barton

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

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

google-analytics-data>=0.17.0,<1.0.0
python-dotenv>=1.0.0,<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)


def _safe_output_path(path: str) -> str:
    """Validate output path does not escape working directory."""
    resolved = os.path.realpath(path)
    cwd = os.path.realpath(os.getcwd())
    if not resolved.startswith(cwd + os.sep) and resolved != cwd:
        raise ValueError(f"Output path must be within working directory: {cwd}")
    return resolved


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:
            safe_path = _safe_output_path(args.output)
            with open(safe_path, "w", encoding="utf-8") as f:
                f.write(output)
            print(f"Analysis saved to {args.output}", file=sys.stderr)
        else:
            print(output)

    except ValueError as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)
    except Exception:
        print("Error: Analysis failed. Check credentials and property ID.", 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("Error: Required packages not installed.", file=sys.stderr)
    print("Install with: pip install google-analytics-data python-dotenv", file=sys.stderr)
    sys.exit(1)


def _safe_output_path(path: str) -> str:
    """Validate output path does not escape working directory."""
    resolved = os.path.realpath(path)
    cwd = os.path.realpath(os.getcwd())
    if not resolved.startswith(cwd + os.sep) and resolved != cwd:
        raise ValueError(f"Output path must be within working directory: {cwd}")
    return resolved


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(
                "Failed to initialize Google Analytics client. 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("Failed to run report. Check property ID and service account permissions.")

        # 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:
            safe_path = _safe_output_path(args.output)
            with open(safe_path, "w", encoding="utf-8") as f:
                f.write(output)
            print(f"Report saved to {args.output}", file=sys.stderr)
        else:
            print(output)

    except ValueError as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)
    except Exception:
        print("Error: Failed to run report. Check credentials and property ID.", 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()