plausible-stats

#!/usr/bin/env python3

"""
Load site data from Plausible Analytics API and print to stdout.

Input is a TSV file containing the name of the site and its Plausible domain
"""
import argparse
import datetime
import json
import os
import sys
from calendar import month
from collections import OrderedDict
import csv
from typing import List

import requests

try:
    PLAUSIBLE_KEY = os.environ['PLAUSIBLE_KEY']
except KeyError:
    print("Please set PLAUSIBLE_KEY environment variable")
    sys.exit(1)

FIELDS = ["visitors", "visits", "pageviews", "views_per_visit", "bounce_rate", "visit_duration"]
START = '2023-06-01'


def geo(args, sites: OrderedDict):
    """Load geo data"""
    session = requests.session()
    session.headers = {'Authorization': "Bearer " + PLAUSIBLE_KEY}

    data = OrderedDict()
    for short_name, domain in sites.items():
        name = f"{short_name} Geo"
        # parse the month of the given start date:
        start = datetime.datetime.strptime(args.start, "%Y-%m-01")
        # get a list of months from start to the current date:
        months = [start + datetime.timedelta(days=31 * i) for i in range(0, (datetime.date.today().year - start.year) * 12 + datetime.date.today().month - start.month + 1)]
        # get the data for each month:
        data[name] = OrderedDict()

        first_of_months = [m.strftime("%Y-%m-01") for m in months]
        country_data = OrderedDict()

        # in order the get all available countries we need to collect all the
        for i, fom in enumerate(first_of_months):
            r = session.get('https://plausible.io/api/v1/stats/breakdown', params=dict(
                site_id=domain,
                period='month',
                property='visit:country',
                date=fom
            ))
            results = r.json()['results']

            for result in results:
                country_code = result['country']
                visitors = result['visitors']
                if country_code not in country_data:
                    country_data[country_code] = {fom: visitors}
                else:
                    country_data[country_code][fom] = visitors

        # reformat the results:
        data[name] = [["country"] + [m.strftime("%Y-%m") for m in months]]
        for country_code, visitors in country_data.items():
            data[name].append([country_code] + [visitors.get(fom, 0) for fom in first_of_months])

    json.dump(data, fp=sys.stdout, indent=2)


def visits(args, sites: OrderedDict):
    """Load all site data"""
    session = requests.session()
    session.headers = {'Authorization': "Bearer " + PLAUSIBLE_KEY}

    data = OrderedDict()
    for name, domain in sites.items():
        r = session.get('https://plausible.io/api/v1/stats/timeseries', params=dict(
            site_id=domain,
            period='custom',
            metrics=','.join(FIELDS),
            date=args.start + "," + datetime.date.today().isoformat()
        ))
        data[name] = [["date"] + FIELDS]
        data[name].extend([[i["date"]] + [i[field] for field in FIELDS] for i in r.json()['results']])

    json.dump(data, fp=sys.stdout, indent=2)


def totals(args, sites: OrderedDict):
    """Load all site data"""
    session = requests.session()
    session.headers = {
        'Authorization': "Bearer " + PLAUSIBLE_KEY,
        'Content-Type': 'application/json'
    }

    # Parse the start date and convert to first of that year
    start = datetime.datetime.strptime(args.start, "%Y-%m-%d").date()
    start_date = start.replace(month=1, day=1)
    end_date = start.replace(month=12, day=31)

    data = OrderedDict()

    metrics = ["visitors", "pageviews"]
    filters = ["contains", "event:page", args.page] if args.page else []

    date_ranges = [(year, [start_date.replace(year=year).strftime('%Y-%m-%d'),
               end_date.replace(year=year).strftime('%Y-%m-%d')])
              for year in range(start.year, datetime.date.today().year + 1)]
    if args.months:
        date_ranges = [(d, [d.strftime('%Y-%m-%d'),
                            last_day_of_month(d).strftime('%Y-%m-%d')])
                       for d in get_monthly_dates(start_date)]

    for period, ranges in date_ranges:
        for name, domain in sites.items():
            payload = dict(
                site_id=domain,
                # period='custom',
                filters=[filters],
                metrics=metrics,
                date_range=ranges,
                include=dict(imports=True)
            )
            # print(json.dumps(payload), file=sys.stderr)
            r = session.post('https://plausible.io/api/v2/query', json=payload)

            # print(r.status_code, file=sys.stderr)
            if name not in data:
                data[name] = []
            apidata = r.json()
            if 'error' in apidata:
                print("Error from Plausible API: " + apidata['error'], file=sys.stderr)
                sys.exit(2)
            results = r.json()['results'][0]['metrics'][0:len(metrics)]
            data[name].append([period] + results)

    if args.format == "csv":
        import csv
        writer = csv.writer(sys.stdout)
        writer.writerow(['site', 'month' if args.months else 'year'] + metrics)
        for site, metrics in data.items():
            for row in metrics:
                writer.writerow([site] + row)
    else:
        json.dump(data, fp=sys.stdout, indent=2)


def last_day_of_month(first_day):
    """
    Get the last day of the month given the first day using only standard library.
    """
    # Calculate the year and month of the next month
    year = first_day.year + (first_day.month // 12)
    month = (first_day.month % 12) + 1

    # First day of next month
    next_month = datetime.date(year, month, 1)

    # Subtract one day to get the last day of the original month
    last_day = next_month - datetime.timedelta(days=1)
    return last_day


def get_monthly_dates(start_date):
    """
    Generate a list of dates, one per month,
    from the start date until today.
    """
    # Get today's date
    today = datetime.date.today().replace(day=1)

    # Initialize result list and current date
    dates = []
    current = start_date.replace(day=1)

    # Generate dates until we reach or exceed today
    while current <= today:
        dates.append(current)

        # Move to the first day of the next month
        year = current.year + (current.month // 12)
        month = (current.month % 12) + 1
        current = datetime.date(year, month, 1)  # Always use day 1

    return dates


if __name__ == "__main__":
    # Parse arguments
    parser = argparse.ArgumentParser(description='Load site data from Plausible Analytics API and print to stdout.')
    parser.add_argument('-g', '--geo', action='store_true', help="Extract geospatial data for months to date")
    parser.add_argument('-p', '--page', nargs='*',  help="Count unique visitors for one or more matching "
                                                         "page prefixes, e.g. /units/de-002302")
    parser.add_argument('-s', '--start', default=START, help="Start date for data as YYYY-MM-DD")
    parser.add_argument('-t', '--totals', action='store_true', help="Extract totals data for years to date")
    parser.add_argument('-S', '--site', nargs='*', help='A specific site to query')
    parser.add_argument('-f', '--format', default="json", help="Output format: either json or csv")
    parser.add_argument('-m', '--months', action="store_true", help="Output date in month (not year) ranges")
    args = parser.parse_args()

    # Load SITES from TSV file passed via stdin
    sites = OrderedDict()
    if args.site:
        for i, site in enumerate(args.site):
            sites["Site {}".format(i + 1)] = site
    else:
        for row in csv.reader(sys.stdin, delimiter="\t"):
            sites[row[0]] = row[1]

    if args.geo:
        geo(args, sites)
    elif args.totals:
        totals(args, sites)
    else:
        visits(args, sites)