#!/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, stderrCopy=False):  # 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)
        if stderrCopy:
            print(msg, file=sys.stderr)


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 = "%04d-%02d" % (currentYear, currentMonth)
    currentMonth -= 1
    if currentMonth == 0:
        currentMonth = 12
        currentYear -= 1
    months.append(date)

async def get_stats(client, listpart, domainpart, date):
    """
    Get stats data
    """
    url = f"https://lists.apache.org/api/stats.json?list={listpart}&domain={domainpart}&d={date}&quick"
    res = None
    mldata = None
    try:
        res = await client.get(url)
        ct = res.content_type
        if ct == 'application/json':
            mldata = await res.json()
        else:
            tsprint(f"Unexpected Content-Type {ct} fetching {url}", stderrCopy=True)
            tsprint(await res.text())
            raise TypeError(f"Unexpected Content-Type {ct}")
    # syncio.exceptions.TimeoutError is aliased to TimeoutError since Python 3.11
    # so this should work for both
    except asyncio.exceptions.TimeoutError:
        tsprint(f"Timeout fetching {url}", stderrCopy=True) # log the error
        raise
    except Exception as err: # anything other than Timeout or Content Type
        tsprint(f"Failed fetching {url}: {repr(err)}", stderrCopy=True)
        if hasattr(err, 'add_note'):
            err.add_note(f"Failed fetching {url}: {repr(err)}")
        else:
            raise RuntimeError(f"Failed fetching {url}") from err
        raise
    return mldata

async def get_stats_retry(client, listpart, domainpart, date):
    """Get stats, retrying for some errors"""
    MAX_RETRIES = 2
    retries = 0
    while True:
        try:
            mldata = await get_stats(client, listpart, domainpart, date)
            return mldata, retries
        # syncio.exceptions.TimeoutError is aliased to TimeoutError since Python 3.11
        # so this should work for both
        except (asyncio.exceptions.TimeoutError, TypeError) as e:
            retries += 1
            if retries > MAX_RETRIES:
                raise
            tsprint(f"Caught error {repr(e)}, will retry",  stderrCopy=True)
            time.sleep(20*retries)
            tsprint("Retrying ...", stderrCopy=True)

async def weekly_stats(client, listpart, domainpart, date):
    """
    Read the weekly stats
    """
    project = domainpart.replace(".apache.org", "")
    if not project:
        return {}
    mldata, retries = await get_stats_retry(client, listpart, domainpart, date)
    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('Should not happen?', repr(err), stderrCopy=True)
    return count, weekly, retries


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

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
    retries = 0
    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:
                mldatat, retryct = await get_stats_retry(client, listpart, domainpart, '1990-01')
                active_months = mldatat['active_months']
                retries += retryct
                recent = list([x, active_months.get(x)] for x in months)
                total = sum((x[1] or 0) for x in recent)
                tsprint(f"Recent emails for {listpart}@{domainpart}: {recent} total: {total}")
                fetch = list(filter(lambda x: active_months.get(x,0) > 0, months))
                if len(fetch) > 0:
                    tsprint(f"Fetching data for {fetch}")
                    results = await asyncio.gather(*[weekly_stats(client, listpart, domainpart, date) for date in fetch])
                else:
                    tsprint("** No data to fetch **")
                    results = []
            for result in results:
                ct, weeks, r = result
                retries += r
                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"Failed after total {retries} retries", stderrCopy=True)
            tsprint(f"While processing {listpart}@{domainpart} failed with {repr(e)} ", stderrCopy=True)
            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 %d mails (%d 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:
        file = __MAILDATA_EXTENDED
        tsprint(f"Completed scanning, writing JSON file {file}")
        with open(file, "w+", encoding='utf-8') as f:
            json.dump(mls, f, indent=1, sort_keys=True)

        tsprint(f"Dumped JSON file")
    return retries


if __name__ == "__main__":
    try:
        tsprint("Started")
        # Get the index of mailing lists
        url = "https://lists.apache.org/api/preferences.json"
        try:
            data = requests.get(url, timeout=30).json()
        except requests.exceptions.Timeout:
            tsprint(f"Timeout fetching {url}", stderrCopy=True) # log the error
            data = requests.get(url, timeout=30).json()
        tsprint("Fetched %d lists in preferences data" % len(data.get('lists',())))
        retries = asyncio.run(gather_stats())
        elapsed = time.time() - startTime
        msg = "Completed in %d seconds, retries %d" % (elapsed, retries)
        tsprint(msg)
        if retries > 0: # send alert that retry occurred
            print(msg, file=sys.stderr)            
    except Exception as e:
        tsprint("Failed: %s" % repr(e))
        raise # ensure it alerts the mailing list