opencode-skill/skills/umami/scripts/umami_client.py

#!/usr/bin/env python3
"""
Umami Analytics Client

Self-hosted Umami integration with username/password authentication.
Creates websites, gets tracking codes, and fetches analytics data.

Automatically loads credentials from:
- ~/.config/opencode/.env (unified .env)
- .env (local, for development)
"""

import os
import sys
import requests
import argparse
from datetime import datetime, timedelta
from typing import Dict, Optional, List
from pathlib import Path
from skills._env_loader import load_unified_env

load_unified_env()


class UmamiClient:
    """Umami Analytics API client with username/password auth"""

    def __init__(
        self,
        umami_url: str = None,
        username: str = None,
        password: str = None,
        token: str = None,
    ):
        """
        Initialize Umami client

        Args:
            umami_url: Umami instance URL (loaded from .env if not provided)
            username: Umami username/email (loaded from .env if not provided)
            password: Umami password (loaded from .env if not provided)
            token: Bearer token (optional, if already have)
        """
        # Load from environment if not provided
        self.umami_url = umami_url or os.getenv("UMAMI_URL", "")
        self.username = username or os.getenv("UMAMI_USERNAME", "")
        self.password = password or os.getenv("UMAMI_PASSWORD", "")
        self.token = token
        self.user_id = None

        # Auto-login if credentials provided
        if self.username and self.password and not token:
            self.login()

    def login(self) -> Dict:
        """Login to Umami and get bearer token"""
        try:
            url = f"{self.umami_url}/api/auth/login"
            data = {"username": self.username, "password": self.password}

            response = requests.post(url, json=data)
            response.raise_for_status()
            result = response.json()

            if "token" in result:
                self.token = result["token"]
                self.user_id = result.get("user", {}).get("id")

                return {
                    "success": True,
                    "token": self.token,
                    "user_id": self.user_id,
                    "username": result.get("user", {}).get("username"),
                }
            else:
                return {"success": False, "error": "No token in response"}

        except Exception as e:
            return {"success": False, "error": str(e)}

    def _get_headers(self) -> Dict:
        """Get request headers with auth"""
        if not self.token:
            if self.username and self.password:
                self.login()

        return {
            "Authorization": f"Bearer {self.token}",
            "Content-Type": "application/json",
            "Accept": "application/json",
        }

    def create_website(self, name: str, domain: str) -> Dict:
        """Create new Umami website"""
        try:
            url = f"{self.umami_url}/api/websites"
            data = {"name": name, "domain": domain}

            response = requests.post(url, json=data, headers=self._get_headers())
            response.raise_for_status()
            result = response.json()

            return {
                "success": True,
                "website_id": result.get("id"),
                "name": result.get("name"),
                "domain": result.get("domain"),
                "created_at": result.get("createdAt"),
                "tracking_url": f"{self.umami_url}/script.js",
                "tracking_script": self._get_tracking_script(result.get("id")),
            }

        except Exception as e:
            return {"success": False, "error": str(e)}

    def get_website_by_domain(self, domain: str) -> Optional[Dict]:
        """Find website by domain"""
        try:
            websites = self.list_websites()
            for site in websites:
                if domain in site.get("domain", ""):
                    return site
            return None
        except:
            return None

    def list_websites(self) -> List[Dict]:
        """Get all websites"""
        try:
            url = f"{self.umami_url}/api/websites"
            response = requests.get(url, headers=self._get_headers())
            response.raise_for_status()
            result = response.json()

            # Handle both array and paginated response
            if isinstance(result, list):
                return result
            elif "data" in result:
                return result["data"]
            else:
                return []

        except Exception as e:
            print(f"Error listing websites: {e}")
            return []

    def get_stats(self, website_id: str, days: int = 30) -> Dict:
        """Get website statistics"""
        try:
            end_date = datetime.now()
            start_date = end_date - timedelta(days=days)

            url = f"{self.umami_url}/api/websites/{website_id}/stats"
            params = {
                "startAt": int(start_date.timestamp() * 1000),
                "endAt": int(end_date.timestamp() * 1000),
            }

            response = requests.get(url, headers=self._get_headers(), params=params)
            response.raise_for_status()
            stats = response.json()

            return {
                "success": True,
                "website_id": website_id,
                "period": f"last_{days}_days",
                "pageviews": stats.get("pageviews", 0),
                "uniques": stats.get("uniques", 0),
                "bounces": stats.get("bounces", 0),
                "totaltime": stats.get("totaltime", 0),
                "avg_session_duration": stats.get("totaltime", 0)
                / max(stats.get("visits", 1), 1),
                "bounce_rate": stats.get("bounces", 0)
                / max(stats.get("visits", 1), 1)
                * 100,
            }

        except Exception as e:
            return {"success": False, "error": str(e)}

    def _get_tracking_script(self, website_id: str) -> str:
        """Generate tracking script HTML"""
        return f'<script defer src="{self.umami_url}/script.js" data-website-id="{website_id}"></script>'

    def add_tracking_to_astro(self, website_repo: str, website_id: str) -> Dict:
        """Add Umami tracking to Astro website"""
        try:
            tracking_script = self._get_tracking_script(website_id)

            # Find Astro layout file
            layout_paths = [
                os.path.join(website_repo, "src/layouts/Layout.astro"),
                os.path.join(website_repo, "src/layouts/BaseHead.astro"),
                os.path.join(website_repo, "src/pages/_document.tsx"),
                os.path.join(website_repo, "src/app.html"),
            ]

            layout_file = None
            for path in layout_paths:
                if os.path.exists(path):
                    layout_file = path
                    break

            if not layout_file:
                layouts_dir = os.path.join(website_repo, "src/layouts")
                if os.path.exists(layouts_dir):
                    for f in os.listdir(layouts_dir):
                        if f.endswith(".astro"):
                            layout_file = os.path.join(layouts_dir, f)
                            break

            if not layout_file:
                return {"success": False, "error": "No Astro layout file found"}

            with open(layout_file, "r", encoding="utf-8") as f:
                content = f.read()

            if "</head>" in content:
                content = content.replace(
                    "</head>", f"    {tracking_script}\n  </head>"
                )
            else:
                content += f"\n{tracking_script}\n"

            with open(layout_file, "w", encoding="utf-8") as f:
                f.write(content)

            return {"success": True, "layout_file": layout_file, "tracking_added": True}

        except Exception as e:
            return {"success": False, "error": str(e)}


def main():
    """Main CLI entry point"""
    parser = argparse.ArgumentParser(description="Umami Analytics Client")

    parser.add_argument(
        "--action",
        required=True,
        choices=[
            "create-website",
            "get-tracking",
            "add-tracking",
            "get-stats",
            "list-websites",
        ],
    )
    parser.add_argument(
        "--umami-url", help="Umami instance URL (or set UMAMI_URL in .env)"
    )
    parser.add_argument(
        "--username", help="Umami username (or set UMAMI_USERNAME in .env)"
    )
    parser.add_argument(
        "--password", help="Umami password (or set UMAMI_PASSWORD in .env)"
    )
    parser.add_argument("--website-name", help="Website name (for create)")
    parser.add_argument("--website-domain", help="Website domain (for create/find)")
    parser.add_argument("--website-id", help="Website ID (for stats)")
    parser.add_argument(
        "--website-repo", help="Path to website repo (for add-tracking)"
    )
    parser.add_argument("--days", type=int, default=30, help="Days for stats")

    args = parser.parse_args()

    print(f"\n📊 Umami Analytics Client")
    print(f"URL: {args.umami_url or os.getenv('UMAMI_URL', 'Not set')}\n")

    # Initialize client (loads credentials from .env automatically)
    client = UmamiClient(
        umami_url=args.umami_url, username=args.username, password=args.password
    )

    if args.action == "create-website":
        if not args.website_name or not args.website_domain:
            print("Error: --website-name and --website-domain required")
            return

        print(f"Creating website: {args.website_name} ({args.website_domain})")
        result = client.create_website(args.website_name, args.website_domain)

        if result["success"]:
            print(f"\n✅ Website created!")
            print(f"  ID: {result['website_id']}")
            print(f"  Name: {result['name']}")
            print(f"  Domain: {result['domain']}")
            print(f"  Tracking: {result['tracking_url']}")
            print(f"\nScript:\n{result['tracking_script']}")
        else:
            print(f"\n❌ Failed: {result['error']}")

    elif args.action == "get-tracking":
        if not args.website_id:
            print("Error: --website-id required")
            return

        script = client._get_tracking_script(args.website_id)
        print(f"\nTracking script for {args.website_id}:")
        print(script)

    elif args.action == "add-tracking":
        if not args.website_id or not args.website_repo:
            print("Error: --website-id and --website-repo required")
            return

        print(f"Adding tracking to: {args.website_repo}")
        result = client.add_tracking_to_astro(args.website_repo, args.website_id)

        if result["success"]:
            print(f"\n✅ Tracking added!")
            print(f"  Layout: {result['layout_file']}")
        else:
            print(f"\n❌ Failed: {result['error']}")

    elif args.action == "get-stats":
        if not args.website_id:
            print("Error: --website-id required")
            return

        print(f"Getting stats for last {args.days} days...")
        stats = client.get_stats(args.website_id, args.days)

        if stats["success"]:
            print(f"\n📊 Analytics ({stats['period']}):")
            print(f"  Pageviews: {stats['pageviews']:,}")
            print(f"  Unique visitors: {stats['uniques']:,}")
            print(f"  Bounces: {stats['bounces']:,}")
            print(f"  Bounce rate: {stats['bounce_rate']:.1f}%")
            print(f"  Avg session: {stats['avg_session_duration']:.1f}s")
        else:
            print(f"\n❌ Failed: {stats['error']}")

    elif args.action == "list-websites":
        print("Listing websites...")
        websites = client.list_websites()

        print(f"\nFound {len(websites)} websites:")
        for site in websites:
            print(f"  • {site.get('name')} - {site.get('domain')}")


if __name__ == "__main__":
    main()