scuti/xdfcgi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

renamed function 'i' and 'duplicatestest' for readability

Browse Source
This commit is contained in:
scuti 2024-10-28 23:24:43 -07:00
parent 3834385d2f
commit 70fb324923
1 changed files with 36 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
scripts/import-from-xon.py
View File

@ -124,7 +124,9 @@ def filters(db):
# Functions: Database Creation # Functions: Database Creation
#------------------------------------------------+ #------------------------------------------------+
def inserttodb(c, q, d): # insert new data directly into new database file
def insert_to_database(d, s):
def insert(c, q, d):
for x in d: for x in d:
# possible to do executemany # possible to do executemany
# but want to be able to catch the problematic rows # but want to be able to catch the problematic rows
@ -135,29 +137,30 @@ def inserttodb(c, q, d):
except sql.ProgrammingError as e: except sql.ProgrammingError as e:
print(e) print(e)
print(x) print(x)
return
#------------------------------------------------+
# insert new data directly into new database file
def i(d, s):
con = sql.connect(d) con = sql.connect(d)
with con: with con:
csr = con.cursor() csr = con.cursor()
try: try:
times, ranks, ids, speed, speed_ids = filters(get_list_from_server_txt(s)) times, \
ranks, \
ids, \
speed, \
speed_ids = filters(get_list_from_server_txt(s))
if times: if times:
inserttodb(csr, "INSERT OR REPLACE INTO Cts_times VALUES(?, ?, ?, ?)", times) insert(csr, "INSERT OR REPLACE INTO Cts_times VALUES(?, ?, ?, ?)", times)
logging.info('\n'.join(y for y in [str(x) for x in times])) logging.info('\n'.join(y for y in [str(x) for x in times]))
if ranks: if ranks:
inserttodb(csr, "INSERT OR REPLACE INTO Cts_ranks VALUES(?, ?, ?, ?)", ranks) insert(csr, "INSERT OR REPLACE INTO Cts_ranks VALUES(?, ?, ?, ?)", ranks)
logging.info('\n'.join(y for y in [str(x) for x in ranks])) logging.info('\n'.join(y for y in [str(x) for x in ranks]))
if ids: if ids:
inserttodb(csr, "INSERT OR REPLACE INTO Id2alias VALUES(?, ?, ?)", ids) insert(csr, "INSERT OR REPLACE INTO Id2alias VALUES(?, ?, ?)", ids)
logging.info('\n'.join(y for y in [str(x) for x in ids])) logging.info('\n'.join(y for y in [str(x) for x in ids]))
if speed: if speed:
inserttodb(csr, "INSERT OR REPLACE INTO Speed VALUES(?, ?)", speed) insert(csr, "INSERT OR REPLACE INTO Speed VALUES(?, ?)", speed)
if speed_ids: if speed_ids:
inserttodb(csr, "INSERT OR REPLACE INTO Fastest_players VALUES(?, ?)", speed_ids) insert(csr, "INSERT OR REPLACE INTO Fastest_players VALUES(?, ?)", speed_ids)
except sql.Error: except sql.Error:
logging.exception("sql error encountered in function 'i'") logging.exception("sql error encountered in function 'i'")
if con: if con:
@ -165,29 +168,29 @@ def i(d, s):
# 'insert' new data into a file i.e sql query file # 'insert' new data into a file i.e sql query file
def write_query(out_file, data): def write_query(out_file, data):
with open(out_file, 'w', encoding='utf-8') as file_handle:
times, \ times, \
ranks, \ ranks, \
ids, \ ids, \
speed, \ speed, \
speed_ids = filters(get_list_from_server_txt(data)) speed_ids = filters(get_list_from_server_txt(data))
with open(out_file, 'w', encoding='utf-8') as file_handle:
for t in times: for t in times:
file_handle.write("INSERT OR REPLACE INTO Cts_times VALUES(\'%s\', \'%s\', %s, %s);\n" % tuple(t)) file_handle.write("INSERT OR REPLACE INTO Cts_times VALUES(\'%s\', \'%s\', %s, %s);\n" % tuple(t))
for r in ranks: for r in ranks:
file_handle.write("INSERT OR REPLACE INTO Cts_ranks VALUES(\'%s\', \'%s\', %s, \'%s\');\n" % tuple(r)) file_handle.write("INSERT OR REPLACE INTO Cts_ranks VALUES(\'%s\', \'%s\', %s, \'%s\');\n" % tuple(r))
for i in ids: for i in ids:
file_handle.write("INSERT OR REPLACE INTO Id2alias VALUES(\'%s\', \'%s\', \'%s\');\n" % tuple(i)) file_handle.write("INSERT OR REPLACE INTO Id2alias VALUES(\'%s\', \'%s\', \'%s\');\n" % tuple(i))
pass return
# Test whether repeat rows are added. # Test whether repeat rows are added.
def duplicatestest(d, s): def check_duplicates(database, data):
c = sql.connect(d) c = sql.connect(database)
p = True p = True
with c: with c:
cs = c.cursor() cs = c.cursor()
try: try:
logging.info("Inserting into database (1/2)") logging.info("Inserting into database (1/2)")
i(d, s) insert_to_database(database, data)
logging.info("Querying (1/2)") logging.info("Querying (1/2)")
cs.execute("SELECT * FROM Cts_times") cs.execute("SELECT * FROM Cts_times")
a = cs.fetchall() a = cs.fetchall()
@ -196,7 +199,7 @@ def duplicatestest(d, s):
cs.execute("SELECT * FROM Id2alias") cs.execute("SELECT * FROM Id2alias")
c = cs.fetchall() c = cs.fetchall()
logging.info("Inserting into database (2/2)") logging.info("Inserting into database (2/2)")
i(d, s) insert_to_database(database, data)
logging.info("Querying (2/2)") logging.info("Querying (2/2)")
cs.execute("SELECT * FROM Cts_times") cs.execute("SELECT * FROM Cts_times")
x = cs.fetchall() x = cs.fetchall()
@ -234,7 +237,7 @@ if __name__ == "__main__":
log_file = init_logging() log_file = init_logging()
print("Writing log to ", log_file) print("Writing log to ", log_file)
if args.test: if args.test:
duplicatestest(args.db, args.src) check_duplicates(args.dest, args.src)
if args.query: if args.query:
try: try:
write_query(args.dest, args.src) write_query(args.dest, args.src)
@ -242,4 +245,4 @@ if __name__ == "__main__":
traceback.print_exc() traceback.print_exc()
print("\n\t Exiting - no file to work with.", file=sys.stderr) print("\n\t Exiting - no file to work with.", file=sys.stderr)
else: else:
i(args.dest, args.src) insert_to_database(args.dest, args.src)