import json import sys import psycopg2 from datetime import datetime from datetime import timedelta from configparser import ConfigParser G_FLAG_FULL_IMPORT = False class ParseConfig(Exception): pass def read_config() -> dict[str, str]: parser = ConfigParser() parser.read("database.ini") db = {} if parser.has_section("import"): params = parser.items("import") for param in params: db[param[0]] = param[1] else: raise ParseConfig( 'Section {0} not found in the {1} file'.format("import", "database.ini")) return db def import14_antag(antag, profile_id, conn) -> None: with conn.cursor() as cur: cur.execute("INSERT into antag (profile_id, antag_name) " "VALUES (%(profile_id)s, %(antag_name)s)", { "profile_id": profile_id, "antag_name": antag }) def import14_trait(trait, profile_id, conn) -> None: with conn.cursor() as cur: cur.execute("INSERT into trait (profile_id, trait_name) " "VALUES (%(profile_id)s, %(trait_name)s)", { "profile_id": profile_id, "trait_name": trait }) def import14_job(job_name, priority, profile_id, conn) -> None: with conn.cursor() as cur: cur.execute("INSERT into job (profile_id, job_name, priority) " "VALUES (%(profile_id)s, %(job_name)s, %(priority)s)", { "profile_id": profile_id, "job_name": job_name, "priority": priority }) def last_slot_profile(preference_id, conn) -> int: with conn.cursor() as cur: cur.execute("select slot from profile " "where preference_id = %(preference_id)s " "order by slot desc " "limit 1", { "preference_id": preference_id }) return cur.fetchone()[0] def import14_profile(slot, profile, preference_id, conn) -> None: _markings = "[" _f = True for v in profile["markings"]: if _f: _markings += '"' + v + '"' _f = False else: _markings += ', "' + v + '"' _markings += "]" with conn.cursor() as cur: cur.execute("INSERT into profile (slot, char_name, age, sex, " "hair_name, hair_color, facial_hair_name, " "facial_hair_color, eye_color, skin_color, " "pref_unavailable, preference_id, clothing, gender, " "backpack, species, markings, flavor_text) " "VALUES (%(slot)s, %(char_name)s, %(age)s, %(sex)s, " "%(hair_name)s, %(hair_color)s, %(facial_hair_name)s, " "%(facial_hair_color)s, %(eye_color)s, %(skin_color)s, " "%(pref_unavailable)s, %(preference_id)s, %(clothing)s, " "%(gender)s, %(backpack)s, %(species)s, %(markings)s, " "%(flavor_text)s) " "RETURNING profile_id", { "slot": slot, "char_name": profile["name"], "age": profile["age"], "sex": profile["sex"], "hair_name": profile["hair_name"], "hair_color": profile["hair_color"], "facial_hair_name": profile["facial_hair_name"], "facial_hair_color": profile["facial_hair_color"], "eye_color": profile["eye_color"], "skin_color": profile["skin_color"], "pref_unavailable": 0, "preference_id": preference_id, "clothing": profile["clothing"], "gender": profile["gender"], "backpack": profile["backpack"], "species": profile["species"], "markings": _markings, "flavor_text": profile["flavor_text"] }) profile_id = cur.fetchone()[0] for antag in profile["antag"]: import14_antag(antag, profile_id, conn) for trait in profile["trait"]: import14_trait(trait, profile_id, conn) for job_name, priority in profile["job"].items(): import14_job(job_name, priority, profile_id, conn) def import14_preference(player, conn) -> int: with conn.cursor() as cur: if G_FLAG_FULL_IMPORT: cur.execute("INSERT into preference (user_id, " "selected_character_slot, admin_ooc_color) " "VALUES (%(user_id)s, %(slot)s, %(ooc_color)s) " "RETURNING preference_id", { "user_id": player["uuid"], "slot": player["preference"]["selected_slot"], "ooc_color": player["preference"]["ooc_color"] }) else: cur.execute("select preference_id from preference " "where user_id = %(uuid)s", { "uuid": player["uuid"], }) return cur.fetchone()[0] def import14_playtime(tracker, timespent, uuid, conn) -> None: time_spent = timedelta(seconds=timespent) with conn.cursor() as cur: cur.execute("select time_spent " "from play_time " "where player_id = %(uuid)s " "and tracker like %(tracker)s", { "uuid": uuid, "tracker": tracker }) row = cur.fetchone() if row is None: cur.execute("INSERT into play_time (player_id, tracker, time_spent) " "VALUES (%(player_id)s, %(tracker)s, %(time_spent)s)", { "player_id": uuid, "tracker": tracker, "time_spent": time_spent }) elif row[0] < time_spent: cur.execute("UPDATE play_time " "set time_spent = %(time_spent)s " "where player_id = %(uuid)s " "and tracker like %(tracker)s", { "uuid": uuid, "tracker": tracker, "time_spent": time_spent }) def import14(player) -> None: global G_FLAG_FULL_IMPORT params = read_config() conn = psycopg2.connect(**params) with conn.cursor() as cur: cur.execute("select exists(select 1 from player " "where user_id = %(uuid)s)", { "uuid": player["uuid"] }) exists = cur.fetchone()[0] if not exists: G_FLAG_FULL_IMPORT = True cur.execute("INSERT into player (user_id, first_seen_time, " "last_seen_user_name, last_seen_time, last_seen_address, " "last_seen_hwid) " "VALUES (%(user_id)s, %(first_seen_time)s, " "%(last_seen_user_name)s, %(last_seen_time)s, " "%(last_seen_address)s, %(last_seen_hwid)s)", { "user_id": player["uuid"], "first_seen_time": datetime.fromtimestamp( player["first_seen_time"]), "last_seen_user_name": player["name"], "last_seen_time": datetime.fromtimestamp( player["last_seen_time"]), "last_seen_address": player["ip"], "last_seen_hwid": bytearray.fromhex(player["hwid"]) }) preference_id = import14_preference(player, conn) slot = 0 if not G_FLAG_FULL_IMPORT: slot = last_slot_profile(preference_id, conn) + 1 for profile in player["profiles"]: import14_profile(slot, profile, preference_id, conn) slot += 1 for tracker, timespent in player["playtime"].items(): import14_playtime(tracker, timespent, player["uuid"], conn) conn.commit() conn.close() if __name__ == '__main__': if len(sys.argv) < 2: print("ERROR: missed json file") exit(1) f = open(sys.argv[1]) _player = json.load(f) f.close() import14(_player)