import datetime
import sqlite3 as sql
import csvdef create_database_table(name: str):return """CREATE TABLE IF NOT EXISTS %s (ID TEXT PRIMARY KEY, name TEXT NOT NULL,mode TEXT NOT NULL,var_type TEXT NOT NULL,coefficient TEXT NOT NULL,len TEXT NOT NULL,V0 TEXT NOT NULL,min INTEGER NOT NULL,max INTEGER NOT NULL,pro INTEGER NOT NULL,date TIMESTAMP NOT NULL
)"""% namedef create_database_mml(name: str):return """CREATE TABLE IF NOT EXISTS %s (ID INTEGER PRIMARY KEY AUTOINCREMENT,command TEXT NOT NULL,args TEXT NOT NULL,bootrun BOOLEAN NOT NULL)"""%namedef create_database_trigger(name:str,update_table:str,backup_table:str):return """CREATE TRIGGER IF NOT EXISTS %s AFTER UPDATE ON %sFOR EACH ROW BEGIN UPDATE %sSET V0 = NEW.name,date = NEW.dateWHERE ID = NEW.ID;END;"""%(name,update_table ,backup_table)def csv_to_values(tb:str,filename:str) :rows=[]with open(filename,'r',encoding='utf-8') as csvfile:reader = csv.reader(csvfile)for row in reader:rows.append(row+[datetime.datetime.now().timestamp()])return rowsif __name__ == "__main__":db = "drurmu.db"con = sql.connect(db)cursor = con.cursor()cursor.execute(create_database_table("rmu"))data = csv_to_values("rmu","config/xtfy-au/AU/rmu.csv")cursor.executemany("INSERT INTO rmu VALUES (?,?,?,?,?,?,?,?,?,?,?) ON CONFLICT(ID) DO UPDATE SET name = ""excluded.name,len=excluded.len,pro=excluded.pro,date=excluded.date,V0=excluded.V0,min=excluded.min,max=excluded.max,mode=excluded.mode;",list(map(lambda x:tuple(x),data)))cursor.execute(create_database_table("backup"))cursor.executemany("INSERT INTO backup VALUES (?,?,?,?,?,?,?,?,?,?,?) ON CONFLICT(ID) DO UPDATE SET name = ""excluded.name,len = excluded.len,pro = excluded.pro,date = excluded.date,V0=excluded.V0,""min=excluded.min,max=excluded.max,mode=excluded.mode;",list(map(lambda x:tuple(x),data)))cursor.execute(create_database_mml("mml"))cursor.execute(create_database_trigger("backup_trigger","rmu","backup"))con.commit()con.close()
自动生成 sql 升级语句
def create_csv_sql(filename:str,tb:str):text= ""for row in csv_to_values(tb,filename):text += ("INSERT INTO %s VALUES(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) ON CONFLICT(ID) DO UPDATE SET ""name=excluded.name,""len=excluded.len,""pro=excluded.pro,""date=excluded.date,""V0=excluded.V0,""min=excluded.min,""max=excluded.max,""mode=excluded.mode;\n") % (tb,row[0],row[1],row[2],row[3],row[4],row[5],row[6],row[7],row[8],row[9],row[10])return text
导出 json
cursor.execute("SELECT * FROM rmu")rows = cursor.fetchall()col_name = [des[0] for des in cursor.description]json_data = json.dumps([dict(zip(col_name,row)) for row in rows],indent=4,ensure_ascii=False)with open("rmu.json",'w',encoding='utf-8') as f:f.write(json_data)