player_exporter/export.py

import json
import sys
import psycopg2
import time
from configparser import ConfigParser


class ParseConfig(Exception):
    pass


def read_config() -> dict[str, str]:
    parser = ConfigParser()
    parser.read("database.ini")

    db = {}
    if parser.has_section("export"):
        params = parser.items("export")
        for param in params:
            db[param[0]] = param[1]
    else:
        raise ParseConfig(
            'Section {0} not found in the {1} file'.format("export",
                                                           "database.ini"))

    return db


def export_player(player_name, conn) -> dict:
    _player = {}

    cur = conn.cursor()
    cur.execute("select p.user_id, p.first_seen_time, p.last_seen_time, "
                "p.last_seen_user_name, p.last_seen_address, "
                "p.last_seen_hwid "
                "from player p "
                "where p.last_seen_user_name like %(player_name)s",
                {"player_name": player_name})
    row = cur.fetchone()

    _player.update({"name": row[3]})
    _player.update({"uuid": row[0]})
    _player.update({"first_seen_time": int(time.mktime(row[1].timetuple()))})
    _player.update({"last_seen_time": int(time.mktime(row[2].timetuple()))})
    _player.update({"ip": row[4]})
    _player.update({"hwid": row[5].hex()})

    cur.close()
    return _player


def export_preference(uuid, conn) -> dict:
    cur = conn.cursor()
    cur.execute("select pr.selected_character_slot, pr.admin_ooc_color "
                "from preference pr "
                "where pr.user_id = %(uuid)s",
                {"uuid": uuid})
    row = cur.fetchone()
    _preference = {"selected_slot": row[0], "ooc_color": row[1]}
    cur.close()

    return _preference


def export_antag(profile_id, conn) -> list[str]:
    _antag = []
    cur = conn.cursor()

    cur.execute("select an.antag_name "
                "from antag an "
                "where an.profile_id = %(profile_id)s "
                "order by an.antag_id asc",
                {"profile_id": profile_id})

    row = cur.fetchone()
    while row is not None:
        _antag.append(row[0])
        row = cur.fetchone()

    cur.close()
    return _antag


def export_trait(profile_id, conn) -> list[str]:
    _trait = []
    cur = conn.cursor()

    cur.execute("select tr.* "
                "from trait tr "
                "where tr.profile_id = %(profile_id)s "
                "order by tr.trait_id asc",
                {"profile_id": profile_id})

    row = cur.fetchone()
    while row is not None:
        _trait.append(row[0])
        row = cur.fetchone()

    cur.close()
    return _trait


def export_job(profile_id, conn) -> dict[str, int]:
    _job = {}
    cur = conn.cursor()

    cur.execute("select j.job_name, j.priority "
                "from job j "
                "where j.profile_id = %(profile_id)s "
                "order by j.job_id asc",
                {"profile_id": profile_id})

    row = cur.fetchone()
    while row is not None:
        _job.update({row[0]: row[1]})
        row = cur.fetchone()

    cur.close()
    return _job


def export_profile(uuid, conn) -> list[dict[str, object]]:
    _profiles = []
    cur = conn.cursor()

    cur.execute("select prof.slot, prof.char_name, prof.age, prof.sex, "
                "prof.hair_name, prof.hair_color, prof.facial_hair_name, "
                "prof.facial_hair_color, prof.eye_color, prof.skin_color, "
                "prof.clothing, prof.gender, prof.backpack, prof.species, "
                "prof.markings, prof.flavor_text, prof.profile_id "
                "from profile prof "
                "inner join preference pref on pref.preference_id = prof.preference_id "
                "where pref.user_id = %(uuid)s "
                "order by prof.slot asc",
                {"uuid": uuid})

    row = cur.fetchone()
    while row is not None:
        _profiles.append({
            "slot": row[0],
            "name": row[1],
            "age": row[2],
            "sex": row[3],
            "hair_name": row[4],
            "hair_color": row[5],
            "facial_hair_name": row[6],
            "facial_hair_color": row[7],
            "eye_color": row[8],
            "skin_color": row[9],
            "clothing": row[10],
            "gender": row[11],
            "backpack": row[12],
            "species": row[13],
            "markings": row[14],
            "flavor_text": row[15],
            "antag": export_antag(row[16], conn),
            "trait": export_trait(row[16], conn),
            "job": export_job(row[16], conn)
        })

        row = cur.fetchone()

    cur.close()
    return _profiles


def export_playtime(uuid, conn) -> dict[str, int]:
    _playtime = {}
    cur = conn.cursor()

    cur.execute("select pt.tracker, pt.time_spent "
                "from play_time pt "
                "where pt.player_id = %(uuid)s "
                "order by pt.play_time_id asc",
                {"uuid": uuid})

    row = cur.fetchone()
    while row is not None:
        _playtime.update({row[0]: row[1].total_seconds()})
        row = cur.fetchone()

    cur.close()
    return _playtime


def export(player_name) -> dict:
    params = read_config()
    conn = psycopg2.connect(**params)

    _player = export_player(player_name, conn)
    _player.update({"preference": export_preference(_player["uuid"], conn)})
    _player.update({"profiles": export_profile(_player["uuid"], conn)})
    _player.update({"playtime": export_playtime(_player["uuid"], conn)})

    conn.close()
    return _player


if __name__ == '__main__':
    if len(sys.argv) < 2:
        print("ERROR: missed player name")
        exit(1)
    print(json.dumps(export(sys.argv[1])))