homelab-brain/homelab-ai-bot/core/matomo_client.py

"""Matomo Analytics API Client — Besucherstatistiken fuer arakavanews.com."""

import requests
from datetime import datetime, timedelta

MATOMO_URL = ""
MATOMO_TOKEN = ""
MATOMO_SITE_ID = "1"


def init(cfg):
    global MATOMO_URL, MATOMO_TOKEN, MATOMO_SITE_ID
    MATOMO_URL = cfg.raw.get("MATOMO_URL", "")
    MATOMO_TOKEN = cfg.raw.get("MATOMO_TOKEN", "")
    MATOMO_SITE_ID = cfg.raw.get("MATOMO_SITE_ID", "1")
    return bool(MATOMO_URL and MATOMO_TOKEN)


def _api(method: str, **params) -> dict | list | None:
    try:
        p = {
            "module": "API",
            "method": method,
            "idSite": MATOMO_SITE_ID,
            "format": "json",
            "token_auth": MATOMO_TOKEN,
        }
        p.update(params)
        resp = requests.get(f"{MATOMO_URL}/index.php", params=p, timeout=15)
        resp.raise_for_status()
        return resp.json()
    except Exception as e:
        return {"error": str(e)}


def get_summary(period: str = "day", date: str = "today") -> dict:
    """Besucher-Zusammenfassung (Visits, Unique, Actions, Bounce, Avg Time)."""
    return _api("VisitsSummary.get", period=period, date=date)


def get_visitor_trend(days: int = 30) -> dict:
    """Tageweise Besucherzahlen der letzten N Tage."""
    return _api("VisitsSummary.get", period="day", date=f"last{days}")


def get_top_pages(period: str = "day", date: str = "today", limit: int = 10) -> list:
    """Meistbesuchte Seiten."""
    result = _api("Actions.getPageUrls", period=period, date=date,
                  filter_limit=limit, flat=1)
    if isinstance(result, dict) and "error" in result:
        return []
    return result if isinstance(result, list) else []


def get_referrers(period: str = "day", date: str = "today", limit: int = 10) -> list:
    """Woher kommen Besucher (Suchmaschinen, Social, Direkt)."""
    result = _api("Referrers.getReferrerType", period=period, date=date,
                  filter_limit=limit)
    if isinstance(result, dict) and "error" in result:
        return []
    return result if isinstance(result, list) else []


def get_countries(period: str = "day", date: str = "today", limit: int = 10) -> list:
    """Besucher nach Laendern."""
    result = _api("UserCountry.getCountry", period=period, date=date,
                  filter_limit=limit)
    if isinstance(result, dict) and "error" in result:
        return []
    return result if isinstance(result, list) else []


def get_devices(period: str = "day", date: str = "today") -> list:
    """Besucher nach Geraetetyp (Desktop, Mobile, Tablet)."""
    result = _api("DevicesDetection.getType", period=period, date=date)
    if isinstance(result, dict) and "error" in result:
        return []
    return result if isinstance(result, list) else []


def format_analytics(period: str = "day", date: str = "today") -> str:
    """Kompakter Analytics-Report fuer den Hausmeister-Bot."""
    lines = []

    summary = get_summary(period, date)
    if isinstance(summary, dict) and "error" not in summary:
        visitors = summary.get("nb_uniq_visitors", 0)
        visits = summary.get("nb_visits", 0)
        actions = summary.get("nb_actions", 0)
        bounce = summary.get("bounce_rate", "?")
        avg_time = summary.get("avg_time_on_site", 0)
        avg_min = int(avg_time) // 60
        avg_sec = int(avg_time) % 60
        lines.append(f"Besucher: {visitors} unique, {visits} visits, {actions} Seitenaufrufe")
        lines.append(f"Bounce Rate: {bounce}, Verweildauer: {avg_min}m {avg_sec}s")
    else:
        return f"Matomo nicht erreichbar: {summary}"

    trend = get_visitor_trend(14)
    if isinstance(trend, dict) and "error" not in trend:
        trend_lines = []
        for date_str, data in sorted(trend.items()):
            if isinstance(data, dict):
                v = data.get("nb_uniq_visitors", 0)
                trend_lines.append(f"  {date_str}: {v} Besucher")
            else:
                trend_lines.append(f"  {date_str}: 0")
        if trend_lines:
            lines.append("\nTrend (14 Tage):")
            lines.extend(trend_lines)

    pages = get_top_pages(period, "today", 5)
    if pages:
        lines.append("\nTop Seiten (heute):")
        for p in pages[:5]:
            label = p.get("label", "?")
            hits = p.get("nb_hits", 0)
            lines.append(f"  {label}: {hits}x")

    referrers = get_referrers(period, "today", 5)
    if referrers:
        lines.append("\nTraffic-Quellen (heute):")
        for r in referrers[:5]:
            label = r.get("label", "?")
            visits = r.get("nb_visits", 0)
            lines.append(f"  {label}: {visits} visits")

    countries = get_countries(period, "today", 5)
    if countries:
        lines.append("\nLaender (heute):")
        for c in countries[:5]:
            label = c.get("label", "?")
            visits = c.get("nb_visits", 0)
            lines.append(f"  {label}: {visits}")

    return "\n".join(lines) if lines else "Keine Daten verfuegbar."


def format_trend(days: int = 30) -> str:
    """Besucherentwicklung ueber N Tage — fuer Trend-Fragen."""
    trend = get_visitor_trend(days)
    if isinstance(trend, dict) and "error" in trend:
        return f"Matomo-Fehler: {trend['error']}"

    lines = [f"Besucherentwicklung (letzte {days} Tage):"]
    total = 0
    day_count = 0
    for date_str, data in sorted(trend.items()):
        if isinstance(data, dict):
            v = data.get("nb_uniq_visitors", 0)
            actions = data.get("nb_actions", 0)
            lines.append(f"  {date_str}: {v} Besucher, {actions} Aufrufe")
            total += v
            day_count += 1
        else:
            lines.append(f"  {date_str}: 0")
            day_count += 1

    if day_count > 0:
        avg = total / day_count
        lines.append(f"\nDurchschnitt: {avg:.0f} Besucher/Tag, Gesamt: {total}")

    return "\n".join(lines)