77 lines
1.9 KiB
Python
77 lines
1.9 KiB
Python
from pathlib import Path
|
|
import sys
|
|
|
|
from sqlalchemy import text
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from app.database import engine # noqa: E402
|
|
|
|
|
|
TARGET_CHARSET = "utf8mb4"
|
|
TARGET_COLLATION = "utf8mb4_unicode_ci"
|
|
|
|
APP_TABLES = [
|
|
"attendance_points",
|
|
"personnel",
|
|
"person_roles",
|
|
"person_attendance_points",
|
|
"products",
|
|
"equipment",
|
|
"work_schedules",
|
|
"work_sessions",
|
|
"work_session_devices",
|
|
"production_reports",
|
|
"production_report_items",
|
|
"report_audit_logs",
|
|
"device_qrcodes",
|
|
"notices",
|
|
"notice_points",
|
|
"reconciliation_ledger_entries",
|
|
]
|
|
|
|
|
|
def quote_identifier(value: str) -> str:
|
|
return f"`{value.replace('`', '``')}`"
|
|
|
|
|
|
def table_exists(conn, table: str) -> bool:
|
|
return bool(conn.execute(text("""
|
|
SELECT COUNT(*)
|
|
FROM information_schema.tables
|
|
WHERE table_schema = DATABASE()
|
|
AND table_name = :table
|
|
AND table_type = 'BASE TABLE'
|
|
"""), {"table": table}).scalar_one())
|
|
|
|
|
|
def main() -> None:
|
|
with engine.begin() as conn:
|
|
database = conn.execute(text("SELECT DATABASE()")).scalar_one()
|
|
if database:
|
|
conn.execute(text(
|
|
f"ALTER DATABASE {quote_identifier(database)} "
|
|
f"CHARACTER SET {TARGET_CHARSET} COLLATE {TARGET_COLLATION}"
|
|
))
|
|
|
|
changed = []
|
|
skipped = []
|
|
for table in APP_TABLES:
|
|
if not table_exists(conn, table):
|
|
skipped.append(table)
|
|
continue
|
|
conn.execute(text(
|
|
f"ALTER TABLE {quote_identifier(table)} "
|
|
f"CONVERT TO CHARACTER SET {TARGET_CHARSET} COLLATE {TARGET_COLLATION}"
|
|
))
|
|
changed.append(table)
|
|
|
|
print(f"collation fixed: {', '.join(changed)}")
|
|
if skipped:
|
|
print(f"skipped missing tables: {', '.join(skipped)}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|