179 lines
4.7 KiB
Python
179 lines
4.7 KiB
Python
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
from config import CONFIG
|
||
from db import db_connection
|
||
|
||
|
||
REQUIRED: dict[str, set[str]] = {
|
||
CONFIG.members_table: {
|
||
"id",
|
||
"uid",
|
||
"username",
|
||
"username_clean",
|
||
"balance",
|
||
"remark",
|
||
},
|
||
CONFIG.withdraw_table: {
|
||
"id",
|
||
"username",
|
||
"username_clean",
|
||
"withdraw_amount",
|
||
"withdraw_time",
|
||
"status",
|
||
"remark",
|
||
},
|
||
CONFIG.withdraw_request_table: {
|
||
"id",
|
||
"withdraw_request_id",
|
||
"username_raw",
|
||
"username_clean",
|
||
"withdraw_amount",
|
||
"request_time",
|
||
"operation",
|
||
"audit_status",
|
||
"remark",
|
||
},
|
||
CONFIG.deposit_table: {
|
||
"id",
|
||
"deposit_id",
|
||
"username",
|
||
"username_clean",
|
||
"deposit_amount",
|
||
"actual_arrival_amount",
|
||
"status",
|
||
"deposit_time",
|
||
},
|
||
}
|
||
|
||
|
||
def extract_column_name(row: Any) -> str:
|
||
"""兼容不同 MySQL/PyMySQL 對欄位名稱大小寫的返回方式。"""
|
||
if isinstance(row, dict):
|
||
for key in (
|
||
"column_name_value",
|
||
"column_name",
|
||
"COLUMN_NAME",
|
||
):
|
||
value = row.get(key)
|
||
if value is not None:
|
||
return str(value)
|
||
|
||
if row:
|
||
return str(next(iter(row.values())))
|
||
|
||
if isinstance(row, (list, tuple)) and row:
|
||
return str(row[0])
|
||
|
||
raise KeyError(f"無法取得欄位名稱：{row!r}")
|
||
|
||
|
||
def fetch_table_columns(cursor: Any, table: str) -> set[str]:
|
||
cursor.execute(
|
||
"""
|
||
SELECT COLUMN_NAME AS column_name_value
|
||
FROM information_schema.COLUMNS
|
||
WHERE TABLE_SCHEMA=%s
|
||
AND TABLE_NAME=%s
|
||
ORDER BY ORDINAL_POSITION
|
||
""",
|
||
(
|
||
CONFIG.db_name,
|
||
table,
|
||
),
|
||
)
|
||
|
||
return {
|
||
extract_column_name(row)
|
||
for row in cursor.fetchall()
|
||
}
|
||
|
||
|
||
def main() -> int:
|
||
print("API接口系統資料庫預檢")
|
||
print("=" * 48)
|
||
print(f"資料庫：{CONFIG.db_name}")
|
||
print(f"唯讀模式：{CONFIG.read_only_mode}")
|
||
print(f"會員表：{CONFIG.members_table}")
|
||
print(f"提款記錄表：{CONFIG.withdraw_table}")
|
||
print(f"提款申請表：{CONFIG.withdraw_request_table}")
|
||
print(f"充值表：{CONFIG.deposit_table}")
|
||
print("-" * 48)
|
||
|
||
failed = False
|
||
|
||
try:
|
||
with db_connection() as conn, conn.cursor() as cursor:
|
||
cursor.execute(
|
||
"""
|
||
SELECT
|
||
DATABASE() AS database_name,
|
||
@@global.read_only AS global_read_only
|
||
"""
|
||
)
|
||
server_info = cursor.fetchone() or {}
|
||
|
||
if isinstance(server_info, dict):
|
||
database_name = (
|
||
server_info.get("database_name")
|
||
or server_info.get("DATABASE_NAME")
|
||
or CONFIG.db_name
|
||
)
|
||
if "global_read_only" in server_info:
|
||
mysql_read_only = server_info["global_read_only"]
|
||
else:
|
||
mysql_read_only = server_info.get("GLOBAL_READ_ONLY")
|
||
else:
|
||
database_name = CONFIG.db_name
|
||
mysql_read_only = "unknown"
|
||
|
||
print(f"目前連線資料庫：{database_name}")
|
||
print(f"MySQL global read_only：{mysql_read_only}")
|
||
print("-" * 48)
|
||
|
||
for table, required_columns in REQUIRED.items():
|
||
columns = fetch_table_columns(cursor, table)
|
||
missing = sorted(required_columns - columns)
|
||
|
||
if not columns:
|
||
failed = True
|
||
print(
|
||
f"[失敗] {table}: "
|
||
"資料表不存在，或目前帳號無 information_schema 讀取權限"
|
||
)
|
||
continue
|
||
|
||
if missing:
|
||
failed = True
|
||
print(
|
||
f"[失敗] {table}: "
|
||
f"共 {len(columns)} 個欄位，缺少 {missing}"
|
||
)
|
||
continue
|
||
|
||
print(
|
||
f"[正常] {table}: "
|
||
f"{len(columns)} 個欄位，必要欄位完整"
|
||
)
|
||
|
||
except Exception as exc:
|
||
print(
|
||
f"[失敗] 資料庫預檢執行錯誤："
|
||
f"{type(exc).__name__}: {exc}"
|
||
)
|
||
return 1
|
||
|
||
print("-" * 48)
|
||
|
||
if failed:
|
||
print("預檢結果：存在資料表或欄位問題")
|
||
return 1
|
||
|
||
print("預檢結果：全部正常")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|