64 lines
2.6 KiB
Python
64 lines
2.6 KiB
Python
#!/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())
|