utils.py

import json
import os, base64
import pymysql


class Config:
    def __init__(self) -> None:
        self.name = "Test Website"
        self.port = 8080
        self.dbconf = {}
        self.app_secret_key = base64.b64encode(os.urandom(24)).decode('utf-8')
        self.team_size = 3  # include captain
        self.form_personal = None
        self.form_team = None
        self.form_work = None
        self.all_keys = ['name', 'port', 'dbconf', 'app_secret_key', 'team_size', 'form_personal', 'form_team', 'form_work']
    
    def load_from_file(self, file_path: str, encoding: str='utf-8') -> None:
        with open(file_path, "r", encoding=encoding) as f:
            config = json.load(f)
            for key in self.all_keys:
                if key in config and config[key] is not None:
                    setattr(self, key, config[key])


def init_sql(config: Config, db_config: dict):
    db = pymysql.connect(**db_config)
    mycur = db.cursor()

    # create tables: users
    attr_table = """id INT AUTO_INCREMENT PRIMARY KEY, 
                    username VARCHAR(255) DEFAULT NULL COLLATE utf8_bin, 
                    password VARCHAR(255) DEFAULT NULL COLLATE utf8_bin,"""
    for item in config.form_personal:
        attr_table += f"{item['name']} VARCHAR(255) DEFAULT NULL,"
    attr_table += """captain_username VARCHAR(255) DEFAULT NULL COLLATE utf8_bin,
                     is_agree BOOLEAN DEFAULT NULL,"""
    create_command = f"CREATE TABLE IF NOT EXISTS users ({attr_table[:-1]})"
    mycur.execute(create_command)

    # create tables: teams
    attr_table = """id INT AUTO_INCREMENT PRIMARY KEY,
                    username VARCHAR(255) DEFAULT NULL COLLATE utf8_bin,"""
    for i in range(1, config.team_size):
        attr_table += f"username{i} VARCHAR(255) DEFAULT NULL COLLATE utf8_bin,"
    for i in range(1, config.team_size):
        attr_table += f"is_agree{i} BOOLEAN DEFAULT NULL,"
    if config.form_team is not None:
        for item in config.form_team:
            attr_table += f"{item['name']} VARCHAR(255) DEFAULT NULL,"  
    create_command = f"CREATE TABLE IF NOT EXISTS teams ({attr_table[:-1]})"
    mycur.execute(create_command)

    # create tables: works
    if config.form_work is not None:
        attr_table = """id INT AUTO_INCREMENT PRIMARY KEY,
                        username VARCHAR(255) DEFAULT NULL COLLATE utf8_bin,"""
        for item in config.form_work:
            attr_table += f"{item['name']} VARCHAR(255) DEFAULT NULL,"  
        attr_table += """promotion_status VARCHAR(255) DEFAULT NULL,
                        final_score INT DEFAULT NULL,""" 
        create_command = f"CREATE TABLE IF NOT EXISTS works ({attr_table[:-1]})"
        mycur.execute(create_command)

    # create tables: notifications
    mycur.execute("CREATE TABLE IF NOT EXISTS notifications (id INT AUTO_INCREMENT PRIMARY KEY, title VARCHAR(255) DEFAULT NULL, content TEXT DEFAULT NULL)")

    # create tables: feedbacks
    mycur.execute("CREATE TABLE IF NOT EXISTS feedbacks (id INT AUTO_INCREMENT PRIMARY KEY, title VARCHAR(255) DEFAULT NULL, content TEXT DEFAULT NULL)")

    db.close()

def reset_sql(db_config: dict):
    db = pymysql.connect(**db_config)
    mycur = db.cursor()
    mycur.execute(f"DROP TABLE IF EXISTS users")
    mycur.execute(f"DROP TABLE IF EXISTS teams")
    mycur.execute(f"DROP TABLE IF EXISTS works")
    mycur.execute(f"DROP TABLE IF EXISTS notifications")
    mycur.execute(f"DROP TABLE IF EXISTS feedbacks")
    db.close()