#!/usr/bin/env python3

"""
   Reads public mailing list data from
   https://lists.apache.org/api/preferences.json
   - listing of public mailboxes
   and from each:
   https://lists.apache.org/api/stats.json?list={listpart}&domain={project}.apache.org&d=yyyy-mm&quick
   - messages per week and per last two rolling quarters (92 days)

   Updates:
   data/maildata_extended.json - output data for display
   data/maildata_partial.json - output data for display
"""
import sys

if sys.hexversion < 0x03000000:
    raise ImportError("This script requires Python 3")
import json
import time
import signal
from datetime import datetime
import traceback
import asyncio
import aiohttp
import requests

SECS_PER_DAY = 86400
SECS_PER_WEEK = 604800

# We assume that the script is run from the scripts subdirectory

__RAO_HOME = "../"

__MAILDATA_EXTENDED = __RAO_HOME + "data/maildata_extended.json"
__MAILDATA_PARTIAL  = __RAO_HOME + "data/maildata_partial.json"
# TODO: could the partial data be used for a restart?

def tsprint(s):  # print with timestamp
    msg = "%s %s" % (time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()), s)
    if isinstance(s, Exception):
        print(msg, file=sys.stderr)
        etype, value, tb = sys.exc_info()
        traceback.print_exception(etype, value, tb)
    else:
        print(msg)


interrupted = False


def handle(signum, _frame):
    """
    Handles signals, e.g. ^C (SIGINT) and kill (SIGTERM)
    Sets the interrupted flag on first signal (graceful shutdown)
    Raises KeyboardInterrupt on second signal (abrupt shutdown)
    """
    global interrupted  # otherwise handler does not set the same variable
    if signum == 2:
        print("")  # ensure newline after ^C
    tsprint("Interrupted with %d" % signum)
    if interrupted:  # second interrupt
        raise KeyboardInterrupt  # not generated by Python because we catch its signal
    interrupted = True


tsprint("Start")
startTime = time.time()

# we don't need to retain history
mls = {}

DTNOW = datetime.now()
currentMonth = DTNOW.month
currentYear = DTNOW.year

NOW = time.time()
after = NOW - (SECS_PER_DAY * 92)
wayafter = NOW - (SECS_PER_DAY * 92 * 2)

months = []
for i in range(0, 7):
    date = "%04u%02u" % (currentYear, currentMonth)
    currentMonth -= 1
    if currentMonth == 0:
        currentMonth = 12
        currentYear -= 1
    months.append(date)

# Get the index of mailing lists
data = requests.get("https://lists.apache.org/api/preferences.json", timeout=10).json()
tsprint("Fetched %u bytes of main data" % len(data))


async def weekly_stats(client, listpart, domainpart, date):
    """
    Read the weekly stats from a mbox file, caching the counts.
    """
    project = domainpart.replace(".apache.org", "")
    if not project:
        return {}
    ldate = date[:4] + "-" + date[4:]  # Convert to lists.a.o format, yyyy-mm
    url = f"https://lists.apache.org/api/stats.json?list={listpart}&domain={domainpart}&d={ldate}&quick"
    try:
        mldata = await (await client.get(url)).json()
    except Exception as err:
        tsprint(f"Failed fetching {url}: {err}")
        if hasattr(err, 'add_note'):
            err.add_note(f"Failed fetching {url}: {err}")
        else:
            raise RuntimeError(f"Failed fetching {url}") from err
        raise
    weekly = {}
    count = 0
    if mldata:  # we have a new/updated file to process
        for email in mldata.get("emails", []):
            timestamp = email["epoch"]
            count += 1
            try:
                rounded = timestamp - (timestamp % SECS_PER_WEEK) + SECS_PER_WEEK
                weekly[rounded] = weekly.get(rounded, 0) + 1
            except Exception as err:
                tsprint(err)
    return count, weekly


def add_weeks(total, add):
    for e in add:
        if e in total:
            total[e] += add[e]
        else:
            total[e] = add[e]


tsprint("Started")

signal.signal(signal.SIGINT, handle)  # This stops Python from raising KeyboardInterrupt
signal.signal(signal.SIGTERM, handle)


async def gather_stats():
    # get all the mailing lists
    mlists = []
    for domainpart, lists in sorted(data["lists"].items()):
        if ".apache.org" in domainpart:  # Only project domains
            mlists.extend([f"{listpart}@{domainpart}" for listpart in lists])

    lastCheckpoint = time.time()  # when output files were last saved
    for mailinglist in mlists:
        listpart, domainpart = mailinglist.split("@", maxsplit=1)
        project = domainpart.replace(".apache.org", "")
        ml = f"{project}-{listpart}"
        start = time.time()

        mls[ml] = {}
        mls[ml]["quarterly"] = [0, 0]
        mls[ml]["weekly"] = {}

        mlct = 0

        try:
            tsprint(f"Processing {listpart}@{domainpart}")
            async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(60)) as client:
                results = await asyncio.gather(*[weekly_stats(client, listpart, domainpart, date) for date in months])
            for result in results:
                ct, weeks = result
                add_weeks(mls[ml]["weekly"], weeks)
                mlct += ct
                for week in weeks:
                    if week >= after:
                        mls[ml]["quarterly"][0] += weeks[week]
                    elif week >= wayafter:
                        mls[ml]["quarterly"][1] += weeks[week]

        except KeyboardInterrupt:  # catch the handlers signal
            tsprint("Interrupted processing of %s" % ml)

        except Exception as e:  # add some context
            tsprint(f"While processing {listpart}@{domainpart} failed with {str(e)}")
            if hasattr(e, 'add_note'):
                e.add_note(f"While processing {listpart}@{domainpart}")
            else:
                raise RuntimeError(f"While processing {listpart}@{domainpart}") from e
            raise

        if interrupted:
            break

        tsprint("Info: %s has %u mails (%u secs)" % (ml, mlct, time.time() - start))  # total for mail group
        now = time.time()
        if now - lastCheckpoint > 120:  # checkpoint every 2 minutes
            lastCheckpoint = now
            tsprint("Creating checkpoint of JSON file")
            with open(__MAILDATA_PARTIAL, "w+", encoding='utf-8') as f:
                json.dump(mls, f, indent=1)  # sort_keys is expensive

    # If the run was interrupted, it's likely that data is incomplete
    # for a mailing list and so not useful even as a checkpoint
    if interrupted:
        tsprint("Did not complete successfully, not writing any output")
    else:
        tsprint("Completed scanning, writing JSON file")
        with open(__MAILDATA_EXTENDED, "w+", encoding='utf-8') as f:
            json.dump(mls, f, indent=1, sort_keys=True)

        tsprint(f"Dumped JSON file")
    elapsed = time.time() - startTime
    tsprint("Completed in %d seconds" % elapsed)


if __name__ == "__main__":
    try:
        asyncio.run(gather_stats())
    except Exception as e:
        tsprint("Failed: %s" % repr(e))
        raise # ensure it alerts the mailing list