135 lines
3.4 KiB
Python
135 lines
3.4 KiB
Python
import json
|
|
import requests
|
|
import sys
|
|
import mysql.connector
|
|
import datetime
|
|
import time
|
|
|
|
from typing import List
|
|
|
|
class Application:
|
|
def __init__(self, username):
|
|
"""
|
|
Class containing application data
|
|
|
|
:param username: player's username
|
|
"""
|
|
self.username = username
|
|
|
|
def to_insert_query(self):
|
|
"""
|
|
Created an SQL insert query from the application
|
|
|
|
:return: SQL query string
|
|
"""
|
|
|
|
return """\
|
|
INSERT INTO minecraft_manager_application
|
|
(username, age, player_type, ever_banned, read_rules, accepted, date)
|
|
VALUES
|
|
("{}", 16, "Imported from Lasagn Whitelist", FALSE, "Yes", TRUE, "{}");
|
|
""".format(self.username, datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
|
|
|
|
def __str__(self):
|
|
return self.username
|
|
|
|
|
|
def get_latest_username(uuid: str) -> str:
|
|
"""
|
|
Gets the latest username of a player from the mojang api
|
|
|
|
:param uuid: player's uuid
|
|
:return: player name
|
|
"""
|
|
|
|
r = requests.get("https://api.mojang.com/user/profiles/{}/names".format(uuid))
|
|
|
|
if r.status_code == 200:
|
|
j = r.json()
|
|
|
|
return j[-1]["name"]
|
|
else:
|
|
raise Exception("Error getting user {}. Code: {}: {}".format(uuid, r.status_code, r.text))
|
|
|
|
|
|
def parse_whitelist(whitelist_path: str) -> List[Application]:
|
|
"""
|
|
Process all the applications from the whitelist file
|
|
|
|
:return: list of applications
|
|
"""
|
|
|
|
# Load json from file
|
|
with open(whitelist_path, "r") as f:
|
|
whitelist = json.load(f)
|
|
|
|
# Parse applications from whitelist data
|
|
applications = []
|
|
for entry in whitelist:
|
|
uuid = entry["uuid"]
|
|
|
|
username = get_latest_username(uuid)
|
|
|
|
app = Application(username)
|
|
|
|
applications.append(app)
|
|
|
|
# Prevent Mojang from rate limiting us
|
|
time.sleep(0.5)
|
|
|
|
return applications
|
|
|
|
|
|
def insert_apps_into_db(apps: List[Application], host: str, user: str, pw: str, mcm_db_name: str):
|
|
"""
|
|
Insert applications into the MCM database
|
|
|
|
:param apps: list of applications to import
|
|
:param host: mysql host
|
|
:param user: mysql user
|
|
:param pw: mysql user password
|
|
:param mcm_db_name: MCM database name
|
|
"""
|
|
|
|
# Open database
|
|
db = mysql.connector.connect(host=host, user=user, password=pw, database=mcm_db_name)
|
|
cursor = db.cursor(buffered=True)
|
|
|
|
for app in apps:
|
|
# Check if the player already has an application
|
|
cursor.execute("SELECT * FROM minecraft_manager_application WHERE username = '{}';".format(app.username))
|
|
|
|
result = cursor.fetchall()
|
|
|
|
# If there is not an existing app, insert the application into the DB
|
|
if len(result) == 0:
|
|
cursor.execute(app.to_insert_query())
|
|
|
|
# Commit and close up
|
|
db.commit()
|
|
cursor.close()
|
|
db.close()
|
|
|
|
|
|
def main():
|
|
if len(sys.argv) < 6:
|
|
print("{} [WHITELIST PATH] [MYSQL HOST] [MYSQL USER] [MYSQL PW] [MCM DB Name]".format(sys.argv[0]))
|
|
quit(0)
|
|
else:
|
|
# Parse args
|
|
whitelist_path = sys.argv[1]
|
|
host = sys.argv[2]
|
|
user = sys.argv[3]
|
|
pw = sys.argv[4]
|
|
mc_db_name = sys.argv[5]
|
|
|
|
# Parse whitelist
|
|
apps = parse_whitelist(whitelist_path)
|
|
|
|
# Insert apps into the database
|
|
insert_apps_into_db(apps, host, user, pw, mc_db_name)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|