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("postgresql"): params = parser.items("postgresql") for param in params: db[param[0]] = param[1] else: raise ParseConfig( 'Section {0} not found in the {1} file'.format("postgresql", "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])))