| |
| |
| |
| import sqlite3 |
| import numpy as np |
| from utility.utils import config_dict as DB_CONFIG |
|
|
|
|
| class db_operate(object): |
| |
| def __init__(self, db_path): |
| self.conn = sqlite3.connect(db_path) |
| self.cursor = self.conn.cursor() |
| |
| def create_table(self, SQL): |
| self.cursor.execute(SQL) |
| self.conn.commit() |
|
|
| def insert_data(self, SQL, data): |
| self.cursor.executemany(SQL, data) |
| self.conn.commit() |
| |
| def update_data(self, table_name, data, condition): |
| self.cursor.execute(f"UPDATE {table_name} SET {data} WHERE {condition}") |
| self.conn.commit() |
| |
| def delete_data(self, table_name, condition): |
| self.cursor.execute(f"DELETE FROM {table_name} WHERE {condition}") |
| self.conn.commit() |
| |
| def query_data(self, SQL_statement): |
| self.cursor.execute(SQL_statement) |
| return self.cursor.fetchall() |
|
|
|
|
| if __name__ == "__main__": |
| db_path = DB_CONFIG["db_path"] |
| TABLE = DB_CONFIG["TABLE"] |
| dboperate = db_operate(db_path) |
| print("建表中...") |
| |
| for table_name in TABLE.keys(): |
| table_field_sql = f"create table if not exists {table_name} (" |
| for idx, filed in enumerate(TABLE[table_name]["field"].items()): |
| if idx == len(TABLE[table_name]["field"].items()) - 1: |
| table_field_sql += filed[0] + f" {filed[1][1]})" |
| else: |
| table_field_sql += filed[0] + f" {filed[1][1]}," |
| |
| dboperate.create_table(table_field_sql) |
| |
| print("插入数据中...") |
| |
| TABLE_Values = DB_CONFIG["TABLE_Values"] |
| for table_name in TABLE.keys(): |
| table_field_sql = f"INSERT INTO {table_name} VALUES (" |
| field_len = len(list(TABLE_Values[table_name]["field"].keys())) |
| insert_slot = ",".join(['?' for i in range(field_len)]) |
| table_field_sql += insert_slot + ")" |
| table_val = np.array(list(TABLE_Values[table_name]["field"].values())).T.tolist() |
| dboperate.insert_data(table_field_sql, table_val) |
