feat: 抽象化SQL操作

app/rank_web.py

@@ -11,11 +11,11 @@ class RankWebsite:
         self.app = app
 
     def get_rank(self, order_by: str = "DESC") -> Optional[List[Tuple]]:
-        cur = self._db.search((f"SELECT UserID, Name, Score, Reputation "
-                               f"FROM user "
-                               f"WHERE IsManager = 0 "
-                               f"ORDER BY Reputation {order_by}, Score {order_by}, UserID {order_by} "
-                               f"LIMIT 200;"))
+        cur = self._db.search(columns=['UserID', 'Name', 'Score', 'Reputation'],
+                              table='user',
+                              where='IsManager=0',
+                              order_by=[('Reputation', order_by), ('Score', order_by), ('UserID', order_by)],
+                              limit=200)
         if cur is None:
             return None
         res = []

equipment/scan_garbage.py

@@ -55,10 +55,7 @@ def write_gid_qr(gid: gid_t, path: str, db: DB) -> Tuple[str, Optional[GarbageBa
 
 
 def write_all_gid_qr(path: str, db: DB, where: str = "") -> List[Tuple[str]]:
-    if len(where) > 0:
-        where = f"WHERE {where}"
-
-    cur = db.search(f"SELECT GarbageID FROM garbage {where};")
+    cur = db.search(columns=["GarbageID"], table="garbage", where=where)
     if cur is None:
         return []
 

equipment/scan_user.py

@@ -63,10 +63,7 @@ def write_uid_qr(uid: uid_t, path: str, db: DB, name="nu") -> Tuple[str, Optiona
 
 
 def write_all_uid_qr(path: str, db: DB, name="nu", where: str = "") -> List[str]:
-    if len(where) > 0:
-        where = f"WHERE {where}"
-
-    cur = db.search(f"SELECT UserID, name FROM user {where};")
+    cur = db.search(columns=["UserID", "Name"], table="user", where=where)
     if cur is None:
         return []
 

sql/base_db.py

@@ -1,4 +1,5 @@
 import abc
+from tool.type_ import List, Union, Optional, Tuple
 
 
 class DBException(Exception):
@@ -18,7 +19,7 @@ class DBBit:
     BIT_1 = b'\x01'
 
 
-class Database(metaclass=abc.ABCMeta):
+class HGSDatabase(metaclass=abc.ABCMeta):
     @abc.abstractmethod
     def __init__(self, host: str, name: str, passwd: str):
         self._host = str(host)
@@ -48,36 +49,61 @@ class Database(metaclass=abc.ABCMeta):
         ...
 
     @abc.abstractmethod
-    def search(self, sql) -> any:
+    def search(self, columns: List[str], table: str,
+               where: Union[str, List[str]] = None,
+               limit: Optional[int] = None,
+               offset: Optional[int] = None,
+               order_by: Optional[List[Tuple[str, str]]] = None):
         """
-        执行SQL语句, 仅SELECT
-        :param sql: SELECT的SQL语句
-        :return: 游标
+        执行 查询 SQL语句
+        :param columns: 列名称
+        :param table: 表
+        :param where: 条件
+        :param limit: 限制行数
+        :param offset: 偏移
+        :param order_by: 排序方式
+        :return:
         """
         ...
 
     @abc.abstractmethod
-    def done(self, sql) -> any:
+    def insert(self, table: str, columns: list, values: Union[str, List[str]]):
         """
-        执行SQL语句, 并提交
-        :param sql: SQL语句
-        :return: 游标
+        执行 插入 SQL语句, 并提交
+        :param table: 表
+        :param columns: 列名称
+        :param values: 数据
+        :return:
         """
         ...
 
     @abc.abstractmethod
-    def done_(self, sql) -> any:
+    def delete(self, table: str, where: Union[str, List[str]] = None):
         """
-        执行SQL语句, 但暂时不提交
-        :param sql: SQL语句
-        :return: 游标
+        执行 删除 SQL语句, 并提交
+        :param table: 表
+        :param where: 条件
+        :return:
         """
         ...
 
     @abc.abstractmethod
-    def done_commit(self):
+    def update(self, table: str, kw: dict[str:str], where: Union[str, List[str]] = None):
         """
-        提交由 done_ 执行的SQL语句
+        执行 更新 SQL语句, 并提交
+        :param table: 表
+        :param kw: 键值对
+        :param where: 条件
         :return:
         """
         ...
+
+
+class HGSCursor(metaclass=abc.ABCMeta):
+    @abc.abstractmethod
+    def fetchone(self):
+        ...
+
+    @abc.abstractmethod
+    def fetchall(self):
+        ...

sql/db.py

@@ -1,4 +1,5 @@
 import conf
+from tool.type_ import List
 
 if conf.database.upper() == 'MYSQL':
     try:
@@ -13,12 +14,8 @@ else:
     raise Exception
 
 
-def search_from_garbage_checker_user(columns, where, db: DB):
-    if len(where) > 0:
-        where = f"WHERE {where} "
-
-    column = ", ".join(columns)
-    cur = db.search(f"SELECT {column} FROM garbage_checker_user {where};")
+def search_from_garbage_checker_user(columns: List[str], where, db: DB):
+    cur = db.search(columns=columns, table='garbage_checker_user', where=where)
     if cur is None:
         return None
     res = cur.fetchall()

sql/garbage.py

@@ -10,7 +10,7 @@ def update_garbage_type(where: str, type_: int, db: DB) -> int:
     if len(where) == 0:
         return -1
 
-    cur = db.done(f"UPDATE garbage SET GarbageType={type_} WHERE {where};")
+    cur = db.update(table="garbage", kw={"GarbageType": str(type_)}, where=where)
     if cur is None:
         return -1
     return cur.rowcount
@@ -20,8 +20,8 @@ def update_garbage_check(where: str, check_: bool, db: DB) -> int:
     if len(where) == 0:
         return -1
 
-    i = 1 if check_ else 0
-    cur = db.done(f"UPDATE garbage SET CheckResult={i} WHERE {where};")
+    i: str = '1' if check_ else '0'
+    cur = db.update(table="garbage", kw={"CheckResult": i}, where=where)
     if cur is None:
         return -1
     return cur.rowcount
@@ -110,11 +110,7 @@ def search_garbage_by_fields(columns, gid, uid, cuid, create_time, use_time, loc
 
 
 def search_from_garbage_view(columns, where: str, db: DB):
-    if len(where) > 0:
-        where = f"WHERE {where} "
-
-    column = ", ".join(columns)
-    cur = db.search(f"SELECT {column} FROM garbage {where};")
+    cur = db.search(columns=columns, table="garbage", where=where)
     if cur is None:
         return None
     res = cur.fetchall()
@@ -125,34 +121,38 @@ def count_garbage_by_time(uid: uid_t, db: DB):
     ti: time_t = time.time()
     start = ti - 3.5 * 24 * 60 * 60  # 前后3.5天
     end = ti + 3.5 * 24 * 60 * 60
-    cur = db.search(f"SELECT GarbageID "
-                    f"FROM garbage_time "
-                    f"WHERE UserID = '{uid}' AND UseTime BETWEEN {mysql_time(start)} AND {mysql_time(end)};")
+    cur = db.search(columns=["GarbageID"],
+                    table="garbage_time",
+                    where=[f"UserID = '{uid}'", f"UseTime BETWEEN {mysql_time(start)} AND {mysql_time(end)}"])
     if cur is None:
         return -1
     return cur.rowcount
 
 
-def __find_garbage(sql: str, res_len, db: DB):
-    cur = db.search(sql)
+def __find_garbage(columns: List[str], table: str, where: str, db: DB):
+    cur = db.search(columns=columns, table=table, where=where)
     if cur is None or cur.rowcount == 0:
         return [None, tuple()]
     assert cur.rowcount == 1
     res = cur.fetchone()
-    assert len(res) == res_len
+    assert len(res) == len(columns)
     return GarbageBag(str(res[0])), res
 
 
 def find_not_use_garbage(gid: gid_t, db: DB) -> Union[GarbageBag, None]:
-    return __find_garbage(f"SELECT GarbageID FROM garbage_n WHERE GarbageID = {gid};", 1, db)[0]
+    return __find_garbage(columns=["GarbageID"],
+                          table="garbage_n",
+                          where=f"GarbageID = {gid}",
+                          db=db)[0]
 
 
 def find_wait_garbage(gid: gid_t, db: DB) -> Union[GarbageBag, None]:
     res: Tuple[int, bytes, str, str, str]
     gb: GarbageBag
-    gb, res = __find_garbage(f"SELECT GarbageID, GarbageType, UseTime, UserID, Location "
-                             f"FROM garbage_c "
-                             f"WHERE GarbageID = {gid};", 5, db)
+    gb, res = __find_garbage(columns=["GarbageID", "GarbageType", "UseTime", "UserID", "Location"],
+                             table="garbage_c",
+                             where=f"GarbageID = {gid}",
+                             db=db)[0]
     if gb is None:
         return None
 
@@ -168,9 +168,11 @@ def find_wait_garbage(gid: gid_t, db: DB) -> Union[GarbageBag, None]:
 def find_use_garbage(gid: gid_t, db: DB) -> Union[GarbageBag, None]:
     res: Tuple[int, bytes, str, str, str, bytes]
     gb: GarbageBag
-    gb, res = __find_garbage(f"SELECT GarbageID, GarbageType, UseTime, UserID, Location, CheckResult, CheckerID "
-                             f"FROM garbage_u "
-                             f"WHERE GarbageID = {gid};", 7, db)
+    gb, res = __find_garbage(columns=["GarbageID", "GarbageType", "UseTime", "UserID", "Location",
+                                      "CheckResult", "CheckerID"],
+                             table="garbage_u",
+                             where=f"GarbageID = {gid}",
+                             db=db)[0]
     if gb is None:
         return None
 
@@ -203,7 +205,9 @@ def find_garbage(gid: gid_t, db: DB) -> Union[GarbageBag, None]:
 
 
 def is_garbage_exists(gid: gid_t, db: DB) -> Tuple[bool, int]:
-    cur = db.search(f"SELECT GarbageID, Flat FROM garbage WHERE GarbageID = {gid};")
+    cur = db.search(columns=["GarbageID", "Flat"],
+                    table="garbage",
+                    where=f"GarbageID = {gid}")
     if cur is None or cur.rowcount == 0:
         return False, 0
     assert cur.rowcount == 1
@@ -225,41 +229,35 @@ def update_garbage(garbage: GarbageBag, db: DB) -> bool:
     gid = garbage.get_gid()
     info = garbage.get_info()
 
+    update_kw = {
+        "Flat": "0",
+        "UserID": "NULL",
+        "UseTime": "NULL",
+        "GarbageType": "NULL",
+        "Location": "NULL",
+        "CheckResult": "NULL",
+        "CheckerID": "NULL"
+    }
+
+    if garbage.is_use():
+        update_kw['Flat'] = "1"
+        update_kw['UserID'] = f"'{info['user']}'"
+        update_kw['UseTime'] = f"{mysql_time(info['use_time'])}"
+        update_kw['GarbageType'] = f"{info['type']}"
+        update_kw['Location'] = f"'{info['loc']}'"
+
     if garbage.is_check()[0]:
-        res = db.done(f"UPDATE garbage SET "
-                      f"Flat = 2,"
-                      f"UserID = '{info['user']}',"
-                      f"UseTime = {mysql_time(info['use_time'])},"
-                      f"GarbageType = {info['type']},"
-                      f"Location = '{info['loc']}',"
-                      f"CheckResult = {info['check']},"
-                      f"CheckerID = '{info['checker']}' "
-                      f"WHERE GarbageID = {gid};")
-    elif garbage.is_use():
-        res = db.done(f"UPDATE garbage SET "
-                      f"Flat = 1,"
-                      f"UserID = '{info['user']}',"
-                      f"UseTime = {mysql_time(info['use_time'])},"
-                      f"GarbageType = {info['type']},"
-                      f"Location = '{info['loc']}',"
-                      f"CheckResult = NULL,"
-                      f"CheckerID = NULL "
-                      f"WHERE GarbageID = {gid};")
-    else:
-        res = db.done(f"UPDATE garbage SET "
-                      f"Flat = 0,"
-                      f"UserID = NULL,"
-                      f"UseTime = NULL,"
-                      f"GarbageType = NULL,"
-                      f"Location = NULL,"
-                      f"CheckResult = NULL,"
-                      f"CheckerID = NULL "
-                      f"WHERE GarbageID = {gid};")
+        update_kw['Flat'] = "2"
+        update_kw['CheckResult'] = f"{info['check']}"
+        update_kw['CheckerID'] = f"'{info['checker']}'"
+
+    res = db.update("garbage", kw=update_kw, where=f"GarbageID = {gid}")
     return res is not None
 
 
 def create_new_garbage(db: DB) -> Optional[GarbageBag]:
-    cur = db.done(f"INSERT INTO garbage(CreateTime, Flat) VALUES ({mysql_time()}, 0);")
+    cur = db.insert(table="garbage", columns=["CreateTime", "Flat"], values=f"{mysql_time()}, 0")
+    print(cur)
     if cur is None:
         return None
     assert cur.rowcount == 1
@@ -268,7 +266,7 @@ def create_new_garbage(db: DB) -> Optional[GarbageBag]:
 
 
 def del_garbage_not_use(gid: gid_t, db: DB) -> bool:
-    cur = db.done(f"DELETE FROM garbage_n WHERE GarbageID = {gid};")
+    cur = db.delete(table="garbage_n", where=f"GarbageID = {gid}")
     if cur is None or cur.rowcount == 0:
         return False
     assert cur.rowcount == 1
@@ -276,7 +274,7 @@ def del_garbage_not_use(gid: gid_t, db: DB) -> bool:
 
 
 def del_garbage_wait_check(gid: gid_t, db: DB) -> bool:
-    cur = db.done(f"DELETE FROM garbage_c WHERE GarbageID = {gid};")
+    cur = db.delete(table="garbage_c", where=f"GarbageID = {gid}")
     if cur is None or cur.rowcount == 0:
         return False
     assert cur.rowcount == 1
@@ -284,7 +282,7 @@ def del_garbage_wait_check(gid: gid_t, db: DB) -> bool:
 
 
 def del_garbage_has_check(gid: gid_t, db: DB) -> bool:
-    cur = db.done(f"DELETE FROM garbage_u WHERE GarbageID = {gid};")
+    cur = db.delete(table="garbage_u", where=f"GarbageID = {gid}")
     if cur is None or cur.rowcount == 0:
         return False
     assert cur.rowcount == 1
@@ -292,64 +290,70 @@ def del_garbage_has_check(gid: gid_t, db: DB) -> bool:
 
 
 def del_garbage(gid, db: DB):
-    cur = db.done(f"DELETE FROM garbage WHERE GarbageID = {gid};")
+    cur = db.delete(table="garbage", where=f"GarbageID = {gid}")
     if cur is None or cur.rowcount == 0:
         return False
     assert cur.rowcount == 1
     return True
 
 
-def del_garbage_where_not_use(where, db: DB) -> int:
-    cur = db.done_(f"DELETE FROM garbage_n WHERE {where};")
+def del_garbage_where_not_use(where: str, db: DB) -> int:
+    cur = db.delete(table="garbage_n", where=where)
     if cur is None:
         return -1
     return cur.rowcount
 
 
-def del_garbage_where_wait_check(where, db: DB) -> int:
-    cur = db.done_(f"DELETE FROM garbage_c WHERE {where};")
+def del_garbage_where_wait_check(where: str, db: DB) -> int:
+    cur = db.delete(table="garbage_c", where=where)
     if cur is None:
         return -1
     return cur.rowcount
 
 
-def del_garbage_where_has_check(where, db: DB) -> int:
-    cur = db.done_(f"DELETE FROM garbage_u WHERE {where};")
+def del_garbage_where_has_check(where: str, db: DB) -> int:
+    cur = db.delete(table="garbage_u", where=where)
     if cur is None:
         return -1
     return cur.rowcount
 
 
 def del_garbage_where_scan_not_use(where, db: DB) -> int:
-    cur = db.done(f"SELECT GarbageID FROM garbage_n WHERE {where};")
+    cur = db.search(columns=["GarbageID"],
+                    table="garbage_n",
+                    where=where)
     if cur is None:
         return -1
     return cur.rowcount
 
 
-def del_garbage_where_scan_wait_check(where, db: DB) -> int:
-    cur = db.done(f"SELECT GarbageID FROM garbage_c WHERE {where};")
+def del_garbage_where_scan_wait_check(where: str, db: DB) -> int:
+    cur = db.search(columns=["GarbageID"],
+                    table="garbage_c",
+                    where=where)
     if cur is None:
         return -1
     return cur.rowcount
 
 
 def del_garbage_where_scan_has_check(where, db: DB) -> int:
-    cur = db.done(f"SELECT GarbageID FROM garbage_u WHERE {where};")
+    cur = db.search(columns=["GarbageID"],
+                    table="garbage_u",
+                    where=where)
     if cur is None:
         return -1
     return cur.rowcount
 
 
 def del_all_garbage(db: DB) -> int:
-    cur = db.done(f"DELETE FROM garbage WHERE 1;")
+    cur = db.delete(table="garbage", where='1')
     if cur is None:
         return -1
     return cur.rowcount
 
 
 def del_all_garbage_scan(db: DB) -> int:
-    cur = db.done(f"SELECT GarbageID FROM garbage WHERE 1;")
+    cur = db.search(columns=["GarbageID"], table="garbage", where="1")
     if cur is None:
         return -1
     return cur.rowcount

sql/mysql_db.py

@@ -3,18 +3,30 @@ import threading
 import traceback
 
 from conf import mysql_url, mysql_name, mysql_passwd
-from .base_db import Database, DBCloseException
+from .base_db import HGSDatabase, DBCloseException, HGSCursor
 from tool.type_ import *
 
 
-class MysqlDB(Database):
+class MySQLCursor(HGSCursor):
+    def __init__(self, cursor: pymysql.cursors.Cursor):
+        self._cursor: pymysql.cursors.Cursor = cursor
+
+    def fetchall(self):
+        return self._cursor.fetchall()
+
+    def fetchone(self):
+        return self._cursor.fetchone()
+
+
+class MysqlDB(HGSDatabase):
     def __init__(self, host: str = mysql_url, name: str = mysql_name, passwd: str = mysql_passwd):
         super(MysqlDB, self).__init__(host, name, passwd)
         try:
             self._db = pymysql.connect(user=self._name, password=self._passwd, host=self._host, database="hgssystem")
         except pymysql.err.OperationalError:
             raise
-        self._cursor = self._db.cursor()
+        self._cursor: pymysql.cursors.Cursor = self._db.cursor()
+        self._mysql_cursor = MySQLCursor(self._cursor)
         self._lock = threading.RLock()
 
     def close(self):
@@ -24,6 +36,7 @@ class MysqlDB(Database):
             self._db.close()
         self._db = None
         self._cursor = None
+        self._mysql_cursor = None
         self._lock = None
 
     def is_connect(self) -> bool:
@@ -31,46 +44,91 @@ class MysqlDB(Database):
             return False
         return True
 
-    def get_cursor(self) -> pymysql.cursors.Cursor:
+    def get_cursor(self) -> HGSCursor:
         if self._cursor is None or self._db is None:
             raise DBCloseException
-        return self._cursor
+        return self._mysql_cursor
 
-    def search(self, sql) -> Union[None, pymysql.cursors.Cursor]:
-        if self._cursor is None or self._db is None:
-            raise DBCloseException
+    def search(self, columns: List[str], table: str,
+               where: Union[str, List[str]] = None,
+               limit: Optional[int] = None,
+               offset: Optional[int] = None,
+               order_by: Optional[List[Tuple[str, str]]] = None):
+        if type(where) is list and len(where) > 0:
+            where: str = " WHERE " + " AND ".join(f"({w})" for w in where)
+        elif type(where) is str and len(where) > 0:
+            where = " WHERE " + where
+        else:
+            where: str = ""
 
-        try:
-            self._lock.acquire()  # 上锁
-            self._cursor.execute(sql)
-        except pymysql.MySQLError:
-            print(f"{sql}")
-            traceback.print_exc()
+        if order_by is None:
+            order_by: str = ""
+        else:
+            by = [f" {i[0]} {i[1]} " for i in order_by]
+            order_by: str = " ORDER BY" + ", ".join(by)
+
+        if limit is None:
+            limit: str = ""
+        else:
+            limit = f" LIMIT {limit}"
+
+        if offset is None:
+            offset: str = ""
+        else:
+            offset = f" OFFSET {offset}"
+
+        columns: str = ", ".join(columns)
+        return self.__search(f"SELECT {columns} FROM {table} {where} {order_by} {limit} {offset};")
+
+    def insert(self, table: str, columns: list, values: Union[str, List[str]]):
+        columns: str = ", ".join(columns)
+        if type(values) is str:
+            values: str = f"({values})"
+        else:
+            values: str = ", ".join(f"{v}" for v in values)
+        return self.__done(f"INSERT INTO {table}({columns}) VALUES {values};")
+
+    def delete(self, table: str, where: Union[str, List[str]] = None):
+        if type(where) is list and len(where) > 0:
+            where: str = " AND ".join(f"({w})" for w in where)
+        elif type(where) is not str or len(where) == 0:  # 必须指定条件
             return None
-        finally:
-            self._lock.release()  # 释放锁
-        return self._cursor
 
-    def done(self, sql) -> Union[None, pymysql.cursors.Cursor]:
+        return self.__done(f"DELETE FROM {table} WHERE {where};")
+
+    def update(self, table: str, kw: dict[str:str], where: Union[str, List[str]] = None):
+        if len(kw) == 0:
+            return None
+
+        if type(where) is list and len(where) > 0:
+            where: str = " AND ".join(f"({w})" for w in where)
+        elif type(where) is not str or len(where) == 0:  # 必须指定条件
+            return None
+
+        kw_list = [f"{key} = {kw[key]}" for key in kw]
+        kw_str = ", ".join(kw_list)
+
+        return self.__done(f"UPDATE {table} SET {kw_str} WHERE {where};")
+
+    def __search(self, sql) -> Union[None, HGSCursor]:
         if self._cursor is None or self._db is None:
             raise DBCloseException
 
         try:
-            self._lock.acquire()
+            self._lock.acquire()  # 上锁
             self._cursor.execute(sql)
         except pymysql.MySQLError:
-            self._db.rollback()
             print(f"{sql}")
             traceback.print_exc()
             return None
         finally:
-            self._db.commit()
-            self._lock.release()
-        return self._cursor
+            self._lock.release()  # 释放锁
+        return self._mysql_cursor
 
-    def done_(self, sql) -> Union[None, pymysql.cursors.Cursor]:
+    def __done(self, sql) -> Union[None, HGSCursor]:
         if self._cursor is None or self._db is None:
             raise DBCloseException
+
         try:
             self._lock.acquire()
             self._cursor.execute(sql)
@@ -80,14 +138,6 @@ class MysqlDB(Database):
             traceback.print_exc()
             return None
         finally:
-            self._lock.release()
-        return self._cursor
-
-    def done_commit(self):
-        try:
-            self._lock.acquire()
             self._db.commit()
-        except pymysql.err:
-            traceback.print_exc()
-        finally:
             self._lock.release()
+        return self._mysql_cursor

sql/user.py

@@ -14,7 +14,7 @@ def update_user_score(where: str, score: score_t, db: DB) -> int:
     if len(where) == 0 or score < 0:
         return -1
 
-    cur = db.done(f"UPDATE user SET Score={score} WHERE {where};")
+    cur = db.update(table="user", kw={"score": score}, where=where)
     if cur is None:
         return -1
     return cur.rowcount
@@ -24,7 +24,7 @@ def update_user_reputation(where: str, reputation: score_t, db: DB) -> int:
     if len(where) == 0 or reputation <= 1 or reputation >= 1000:
         return -1
 
-    cur = db.done(f"UPDATE user SET Reputation={reputation} WHERE {where};")
+    cur = db.update(table="user", kw={"Reputation": reputation}, where=where)
     if cur is None:
         return -1
     return cur.rowcount
@@ -46,11 +46,9 @@ def search_user_by_fields(columns, uid: uid_t, name: uname_t, phone: phone_t, db
 
 
 def search_from_user_view(columns, where: str, db: DB):
-    if len(where) > 0:
-        where = f"WHERE {where} "
-
-    column = ", ".join(columns)
-    cur = db.search(f"SELECT {column} FROM user {where};")
+    cur = db.search(columns=columns,
+                    table="user",
+                    where=where)
     if cur is None:
         return None
     result = cur.fetchall()
@@ -63,7 +61,9 @@ def search_from_user_view(columns, where: str, db: DB):
 
 
 def find_user_by_id(uid: uid_t, db: DB) -> Optional[User]:
-    cur = db.search(f"SELECT UserID, Name, IsManager, Score, Reputation FROM user WHERE UserID = '{uid}';")
+    cur = db.search(columns=["UserID", "Name", "IsManager", "Score", "Reputation"],
+                    table="user",
+                    where=f"UserID = '{uid}'")
     if cur is None or cur.rowcount == 0:
         return None
     assert cur.rowcount == 1
@@ -89,7 +89,9 @@ def find_user_by_name(name: uname_t, passwd: passwd_t, db: DB) -> Optional[User]
 
 
 def is_user_exists(uid: uid_t, db: DB) -> bool:
-    cur = db.search(f"SELECT UserID FROM user WHERE UserID = '{uid}';")
+    cur = db.search(columns=["UserID"],
+                    table="user",
+                    where=f"UserID = '{uid}'")
     if cur is None or cur.rowcount == 0:
         return False
     assert cur.rowcount == 1
@@ -101,20 +103,20 @@ def update_user(user: User, db: DB) -> bool:
         return False
 
     uid = user.get_uid()
-    info: dict = user.get_info()
+    info: Dict[str, str] = user.get_info()
     is_manager = info['manager']
     if is_manager == '1':
-        cur = db.done(f"UPDATE user "
-                      f"SET IsManager = {is_manager} "
-                      f"WHERE UserID = '{uid}';")
+        cur = db.update(table="user",
+                        kw={"IsManager": {is_manager}},
+                        where=f"UserID = '{uid}'")
     else:
         score = info['score']
         reputation = info['reputation']
-        cur = db.done(f"UPDATE user "
-                      f"SET IsManager = {is_manager},"
-                      f"    Score = {score},"
-                      f"    Reputation = {reputation} "
-                      f"WHERE UserID = '{uid}';")
+        cur = db.update(table="user",
+                        kw={"IsManager": {is_manager},
+                            "Score": {score},
+                            "Reputation": {reputation}},
+                        where=f"UserID = '{uid}'")
     return cur is not None
 
 
@@ -133,9 +135,10 @@ def create_new_user(name: Optional[uname_t], passwd: Optional[passwd_t], phone:
     if is_user_exists(uid, db):
         return None
     is_manager = '1' if manager else '0'
-    cur = db.done(f"INSERT INTO user(UserID, Name, IsManager, Phone, Score, Reputation, CreateTime) "
-                  f"VALUES ('{uid}', '{name}', {is_manager}, '{phone}', {conf.default_score}, "
-                  f"{conf.default_reputation}, {mysql_time()});")
+    cur = db.insert(table="user",
+                    columns=["UserID", "Name", "IsManager", "Phone", "Score", "Reputation", "CreateTime"],
+                    values=f"'{uid}', '{name}', {is_manager}, '{phone}', {conf.default_score}, "
+                           f"{conf.default_reputation}, {mysql_time()}")
     if cur is None:
         return None
     if is_manager:
@@ -144,7 +147,7 @@ def create_new_user(name: Optional[uname_t], passwd: Optional[passwd_t], phone:
 
 
 def get_user_phone(uid: uid_t, db: DB) -> Optional[str]:
-    cur = db.done(f"SELECT Phone FROM user WHERE UserID = '{uid}';")
+    cur = db.search(columns=["Phone"], table="user", where=f"UserID = '{uid}'")
     if cur is None or cur.rowcount == 0:
         return None
     assert cur.rowcount == 1
@@ -152,10 +155,11 @@ def get_user_phone(uid: uid_t, db: DB) -> Optional[str]:
 
 
 def del_user(uid: uid_t, db: DB) -> bool:
-    cur = db.search(f"SELECT GarbageID FROM garbage_time WHERE UserID = '{uid}';")  # 确保没有引用
+    cur = db.search(columns=["GarbageID"], table="garbage_time", where=f"UserID = '{uid}'")  # 确保没有引用
+
     if cur is None or cur.rowcount != 0:
         return False
-    cur = db.done(f"DELETE FROM user WHERE UserID = '{uid}';")
+    cur = db.delete(table="user", where=f"UserID = '{uid}'")
     if cur is None or cur.rowcount == 0:
         return False
     assert cur.rowcount == 1
@@ -163,18 +167,17 @@ def del_user(uid: uid_t, db: DB) -> bool:
 
 
 def del_user_from_where_scan(where: str, db: DB) -> int:
-    cur = db.search(f"SELECT UserID FROM user WHERE {where};")
-    print(f"SELECT UserID FROM user WHERE {where};")
+    cur = db.search(columns=["UserID"], table="user", where=where)
     if cur is None:
         return -1
     return cur.rowcount
 
 
 def del_user_from_where(where: str, db: DB) -> int:
-    cur = db.search(f"SELECT GarbageID FROM garbage_time WHERE {where};")  # 确保没有引用
+    cur = db.search(columns=["GarbageID"], table="garbage_time", where=where)  # 确保没有引用
     if cur is None or cur.rowcount != 0:
         return False
-    cur = db.done(f"DELETE FROM user WHERE {where};")
+    cur = db.delete(table="user", where=where)
     if cur is None:
         return -1
     return cur.rowcount
@@ -234,4 +237,3 @@ def creat_auto_user_from_csv(path, db: DB) -> List[User]:
             if user is not None:
                 res.append(user)
     return res
-

tk_ui/ranking.py

@@ -36,11 +36,12 @@ class RankingStationBase(metaclass=abc.ABCMeta):
         """
         limit = self.rank_count * self.limit_n
         offset = self.offset + limit * offset  # offset为0表示不移动, 1表示向前, -1表示向后
-        cur = self._db.search((f"SELECT UserID, Name, Score, Reputation "
-                               f"FROM user "
-                               f"WHERE IsManager = 0 "
-                               f"ORDER BY Reputation DESC, Score DESC "
-                               f"LIMIT {limit} OFFSET {offset};"))
+        cur = self._db.search(columns=['UserID', 'Name', 'Score', 'Reputation'],
+                              table='user',
+                              where='IsManager=0',
+                              order_by=[('Reputation', "DESC"), ('Score', "DESC"), ('UserID', "DESC")],
+                              limit=limit,
+                              offset=offset)
         if cur is None or cur.rowcount == 0:
             return False, []
         self.offset = offset

tk_ui/station.py

@@ -170,11 +170,10 @@ class GarbageStationBase(TkEventMain, metaclass=abc.ABCMeta):
         if order_by != 'ASC' and order_by != 'DESC':
             order_by = 'DESC'
 
-        cur = self._db.search((f"SELECT UserID, Name, Score, Reputation "
-                               f"FROM user "
-                               f"WHERE IsManager = 0 "
-                               f"ORDER BY Reputation {order_by}, Score {order_by} "
-                               f"{limit}"))
+        cur = self._db.search(columns=['UserID', 'Name', 'Score', 'Reputation'],
+                              table='user',
+                              where='IsManager=0',
+                              order_by=[('Reputation', order_by), ('Score', order_by), ('UserID', order_by)])
         if cur is None:
             return []
         return list(cur.fetchall())

tk_ui/station_event.py

@@ -105,11 +105,11 @@ class RankingEvent(StationEventBase):
 
     @staticmethod
     def func(db: DB):
-        cur = db.search((f"SELECT UserID, Name, Score, Reputation "
-                         f"FROM user "
-                         f"WHERE IsManager = 0 "
-                         f"ORDER BY Reputation DESC, Score DESC "
-                         f"LIMIT 20;"))
+        cur = db.search(columns=['UserID', 'Name', 'Score', 'Reputation'],
+                        table='user',
+                        where='IsManager=0',
+                        order_by=[('Reputation', "DESC"), ('Score', "DESC"), ('UserID', "DESC")],
+                        limit=20)
         if cur is None:
             return []
         return list(cur.fetchall())