koma-pretix-parser/parse.py

import json
from pathlib import Path
from pprint import pprint
from typing import Collection, Dict, List, Tuple, Union  # I have just python3.8 here :(


# ids of the questions which represent a size of a merch product
SIZE_IDS = (312, 313, 314, 316, 319, 320, 321, 322, 323, 324)
# ID of wishlist for ewiges Frühstück
BREAKFASTWISHES_ID = 309
# Wie heißt die Athenanas?
ATHENE_ID = 305  # ID of the question of how the logo image is to be named
ATHENE_CUSTOM_ID = 306  # custom naming of the Athenanas
HUMAN_ID = 147  # ticket for a human
MASCOT_ID = 148  # ticket for a mascot
# Rahmenprogramm
RAHMERST_ID = 308
RAHMZWEIT_ID = 317
# Yummy!
ESSEN_ID = 302
# diverse checkboxes
BOOLEAN_IDS = (
    304,  # Deutschlandticket
    307,  # Erstkomatikerin
    310,  # Workshop-Themen
    301,  # Mörderspiel
    311,  # nur Rufname auf Badge
)
# where are U from?
UNI_ID = 300
# name ids
NICKNAME_ID = 298



def query_to_csv(filename: Union[str, Path], relevant_items: Collection[int], columns: List[str]) -> str:
    """Parse a JSON export from pretix and query some columns from some orders.

    :param filename: filename of the JSON export to parse.
    :returns: CSV string with one line per order, containing the requested columns.
    """
    with open(filename, 'r', encoding='utf8') as f:
        data = json.load(f)['event']

    items = {i['id']: i['name'] for i in data['items']}
    count_items = {id_: 0 for id_ in items}
    questions = {q['id']: q['identifier'] for q in data['questions']}
    quest_titles = {q['id']: q['question'] for q in data['questions']}
    quest_answers = {id_: {} for id_ in questions}

    head = ("type", "fullname") + tuple(questions[col] for col in columns)
    res = [head]

    for order in data['orders']:
        if order['status'] == 'c':  # skip cancelled orders
            continue
        for position in order['positions']:
            item_id = position['item']
            if item_id not in relevant_items:
                continue
            # specialcasing custom Athenanas names
            def get_answer(qa, col):
                if col != ATHENE_ID or "Sonstiges" not in qa[col]:
                    return qa.get(col, "")
                return qa.get(ATHENE_CUSTOM_ID, "")

            quest_ans = {a['question'] : a['answer'] for a in position['answers']}
            attendee_name = position.get('attendee_name') or ""  # for mascots, they are "None" but empty strings are prettier
            res.append((items[item_id], attendee_name) + tuple(get_answer(quest_ans, col) for col in columns))

    # hacky sanity check that we do not have to escape csv
    for line in res:
        if any('"' in ans for ans in line if ans):
            raise RuntimeError(f"Illegal CSV char, escape needed in line {line}")

    return "\n".join( ('"' + '","'.join(map(str, line)) + '"') for line in res)

def rahmprog_stats(filename) -> str:
    return query_to_csv(filename, (HUMAN_ID,), (NICKNAME_ID, RAHMERST_ID, RAHMZWEIT_ID))

def moerder_list(filename) -> str:
    moerder_id = 301
    result = query_to_csv(filename, (HUMAN_ID,), (NICKNAME_ID, UNI_ID, moerder_id))
    ret = ""
    for line in result.splitlines():
        if line.endswith('"False"'):
            # does not want to take part
            continue
        ret += line + "\n"
    return ret

def badge_data(filename) -> str:
    pronoun_id = 299
    nickname_only_id = 311
    return query_to_csv(filename, (HUMAN_ID, MASCOT_ID), (NICKNAME_ID, nickname_only_id, UNI_ID, pronoun_id, ATHENE_ID))



def parse(filename: Union[str, Path]) -> Tuple[
    Dict[int, str],
    Dict[int, int],
    Dict[int, str],
    Dict[int, Dict[Tuple[int, str], int]],
    Dict[int, str]
]:
    """Parse a JSON export from pretix.

    :param filename: filename of the JSON export to parse.
    :returns: Five mappings:
        * item_id to item name
        * item_id to order count of this item
        * question_id to question identifier
        * question_id to stats for this question, which identifies an answer by its item_id and actual answer.
            The stats map this identifier to number of occurrences
        * question_id to question text (the actual question shown to the user)
    """
    with open(filename, 'r', encoding='utf8') as f:
        data = json.load(f)['event']

    items = {i['id']: i['name'] for i in data['items']}
    count_items = {id_: 0 for id_ in items}
    questions = {q['id']: q['identifier'] for q in data['questions']}
    quest_titles = {q['id']: q['question'] for q in data['questions']}
    quest_answers = {id_: {} for id_ in questions}

    for order in data['orders']:
        if order['status'] == 'c':  # skip cancelled orders
            continue
        for position in order['positions']:
            item_id = position['item']
            count_items[item_id] += 1
            for answer in position['answers']:
                quest_id = answer['question']
                answer_key = (item_id, answer['answer'])
                quest_answers[quest_id][answer_key] = (
                        quest_answers[quest_id].get(answer_key, 0) + 1
                )
    return items, count_items, questions, quest_answers, quest_titles


def wiki_stats(items: Dict[int, str], count_items: Dict[int, int], questions: Dict[int, str],
               quest_answers: Dict[int, Dict[Tuple[int, str], int]], quest_titles: Dict[int, str]) -> str:
    """Generate stats for the Wiki page.

    Input parameters are output of the parse() function.
    """
    ret = ""
    ret += ('== Anmeldezahlen ==\n'
            '{|class="wikitable sortable"\n'
            '! Posten !! Anzahl\n'
            '|-\n')
    for item_id, item_name in items.items():
        ret += f"|| {item_name} || {count_items.get(item_id)}\n|-\n"
    ret += "|}\n\n"

    ret += ('== Merch-Größenverteilung ==\n'
            '{|class="wikitable sortable"\n'
            '! Größe/Eigenschaft !! Wert !! Anzahl\n'
            '|-\n')
    for size_id in SIZE_IDS:
        for size, count in quest_answers[size_id].items():
            ret += f"|| {questions[size_id]}: || {size[1]} || {count}\n|-\n"
    ret += "|}\n\n"

    ret += ('== Rahmenprogramm ==\n'
            '=== Erstwahlen ===\n'
            '{|class="wikitable sortable"\n'
            '! Programmpunkt !! gewählt\n'
            '|-\n')
    for program, count in quest_answers[RAHMERST_ID].items():
        ret += f"|| {program[1]} || {count}\n|-\n"
    ret += ('|}\n'
            '=== Zweitwahlen ===\n'
            '{|class="wikitable sortable"\n'
            '! Programmpunkt !! gewählt\n'
            '|-\n')
    for program, count in quest_answers[RAHMZWEIT_ID].items():
        ret += f"|| {program[1]} || {count}\n|-\n"
    ret += "|}\n\n"

    ret += ('== Essen ==\n'
            '{|class="wikitable sortable"\n'
            '! Präferenz !! gewählt\n'
            '|-\n')
    for pref, count in quest_answers[ESSEN_ID].items():
        ret += f"|| {pref[1]} || {count}\n|-\n"
    ret += ("|}\n"
            "Unverträglichkeiten/Allergien konnten auch angegeben werden, Antworten beim Pretix-Team erfragen."
            "\n\n")

    ret += '== Wünsche an das ewige Frühstück ==\n'
    for wish in quest_answers[BREAKFASTWISHES_ID]:
        ret += '* ' + wish[1].replace('\r\n', '<br/>').replace('\n', '<br/>') + '\n'
    ret += "\n"

    ret += ('== Checkbox-Abfragen ==\n'
            '{|class="wikitable sortable"\n'
            '! Frage !! Antwort !! Anzahl\n'
            '|-\n')
    for field in BOOLEAN_IDS:
        for answer, count in quest_answers[field].items():
            if answer[1] in ("False", ""):
                continue
            ret += f"|| {quest_titles[field]}: || {answer[1]} || {count}\n|-\n"
    ret += "|}\n\n"

    unis = set()
    for uni in quest_answers[UNI_ID]:
        unis.add(uni[1].split()[-1])
    ret += (f'== anwesende Unis ==\n'
            f'Insgesamt {len(unis)}. (Keine Garantie, dass diese automatisierte'
            f' Erkennung sowie folgende Liste korrekt ist.)\n\n'
            + ', '.join(sorted(unis))
            + '\n\n')

    def prefix(id_):
        return "Mensch sagt:" if id_ == HUMAN_ID else "Maskottchen sagt:" if id_ == MASCOT_ID else ""
    ret += ('== Wie heißt das Ding? ==\n'
            '{|class="wikitable sortable"\n'
            '! Variante !! Stimmen\n'
            '|-\n')
    for variant, count in quest_answers[ATHENE_ID].items():
        if "Sonstiges" not in variant[1]:
            ret += f"|| {prefix(variant[0])} {variant[1]} || {count}\n|-\n"
        elif variant[0] == HUMAN_ID: # do not distinguish between human and mascor answers here
            ret += f"|| {variant[1]} || "
            for custom in quest_answers[ATHENE_CUSTOM_ID]:
                ret += f"{custom[1]}<br/>\n"
            ret = ret[:-6]
            ret += f"\n|-\n"
    ret += "|}\n"

    return ret