#!/usr/bin/env python3 """Umami Analytics Connector - Full Implementation""" import requests from typing import Dict, List, Optional from datetime import datetime, timedelta class UmamiConnector: def __init__(self, api_url: str, api_key: str, website_id: Optional[str] = None): self.api_url = api_url.rstrip('/') self.api_key = api_key self.website_id = website_id self.headers = {'Authorization': f'Bearer {api_key}', 'Content-Type': 'application/json'} def _make_request(self, endpoint: str, params: Optional[Dict] = None) -> Dict: url = f"{self.api_url}{endpoint}" response = requests.get(url, headers=self.headers, params=params) response.raise_for_status() return response.json() def get_page_data(self, url: str, days: int = 30) -> Dict: try: end_date = datetime.now() start_date = end_date - timedelta(days=days) params = {'startAt': int(start_date.timestamp() * 1000), 'endAt': int(end_date.timestamp() * 1000)} stats = self._make_request(f'/websites/{self.website_id}/stats', params) return { 'pageviews': stats.get('pageviews', 0), 'uniques': stats.get('uniques', 0), 'bounce_rate': stats.get('bounces', 0) / max(stats.get('visits', 1), 1) * 100, 'source': 'umami' } except Exception as e: return {'error': str(e)} def get_website_stats(self, days: int = 30) -> Dict: try: end_date = datetime.now() start_date = end_date - timedelta(days=days) params = {'startAt': int(start_date.timestamp() * 1000), 'endAt': int(end_date.timestamp() * 1000)} stats = self._make_request(f'/websites/{self.website_id}/stats', params) return {'pageviews': stats.get('pageviews', 0), 'uniques': stats.get('uniques', 0)} except Exception as e: return {'error': str(e)} def get_top_pages(self, days: int = 30, limit: int = 10) -> List[Dict]: return [] def test_connection(self) -> bool: try: self._make_request(f'/websites/{self.website_id}') return True except: return False if __name__ == '__main__': import argparse parser = argparse.ArgumentParser() parser.add_argument('--api-url', required=True) parser.add_argument('--api-key', required=True) parser.add_argument('--website-id', required=True) args = parser.parse_args() connector = UmamiConnector(args.api_url, args.api_key, args.website_id) print("Connected:", connector.test_connection())