fix: 统一数据库表排序规则

app/database.py

@@ -445,7 +445,10 @@ class DatabaseManager:
             mysql_type = self.TYPE_MAPPING.get(col_type, 'VARCHAR(255)')
             column_defs.append(f'`{col}` {mysql_type}')
         
-        sql = f"CREATE TABLE IF NOT EXISTS `{table_name}` ({', '.join(column_defs)}) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4"
+        sql = (
+            f"CREATE TABLE IF NOT EXISTS `{table_name}` ({', '.join(column_defs)}) "
+            "ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci"
+        )
         
         with self.get_connection() as conn:
             with conn.cursor() as cursor:

app/services/cell_data.py

@@ -396,7 +396,7 @@ class CellDataProcessor:
               `功率` varchar(100) DEFAULT NULL,
               `网络` varchar(20) DEFAULT NULL,
               KEY `CGI` (`CGI`)
-            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
+            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
             """
         )
 
@@ -548,9 +548,17 @@ def copy_celldata_tables_to_capacity(app_config: AppConfig, logger=None) -> int:
                 logger.info("CellData 数据库中没有表，跳过复制")
             return 0
 
+        target_charset, target_collation = _database_charset_and_collation(cap_conn)
         copied = 0
         for table in tables:
-            rows = _copy_one_table(cd_conn, cap_conn, table, logger)
+            rows = _copy_one_table(
+                cd_conn,
+                cap_conn,
+                table,
+                target_charset,
+                target_collation,
+                logger,
+            )
             if rows >= 0:
                 copied += 1
         cap_conn.commit()
@@ -571,7 +579,28 @@ def _list_tables(conn) -> list[str]:
         return [list(row.values())[0] for row in cur.fetchall()]
 
 
-def _copy_one_table(src_conn, dst_conn, table: str, logger=None) -> int:
+def _database_charset_and_collation(conn) -> tuple[str, str]:
+    with conn.cursor() as cur:
+        cur.execute(
+            "SELECT @@character_set_database AS `charset_name`, "
+            "@@collation_database AS `collation_name`"
+        )
+        row = cur.fetchone()
+    charset = str(row["charset_name"])
+    collation = str(row["collation_name"])
+    if not re.fullmatch(r"[A-Za-z0-9_]+", charset) or not re.fullmatch(r"[A-Za-z0-9_]+", collation):
+        raise RuntimeError("目标数据库字符集或排序规则名称无效")
+    return charset, collation
+
+
+def _copy_one_table(
+    src_conn,
+    dst_conn,
+    table: str,
+    target_charset: str,
+    target_collation: str,
+    logger=None,
+) -> int:
     with src_conn.cursor() as cur:
         cur.execute(f"SHOW CREATE TABLE `{table}`")
         row = cur.fetchone()
@@ -584,6 +613,10 @@ def _copy_one_table(src_conn, dst_conn, table: str, logger=None) -> int:
     with dst_conn.cursor() as cur:
         cur.execute(f"DROP TABLE IF EXISTS `{table}`")
         cur.execute(create_sql)
+        cur.execute(
+            f"ALTER TABLE `{table}` CONVERT TO CHARACTER SET {target_charset} "
+            f"COLLATE {target_collation}"
+        )
 
     if count == 0:
         if logger: