120 lines
4.1 KiB
Python
120 lines
4.1 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
|
||
"""
|
||
更新数据库表字符集为utf8mb4，以支持emoji和其他特殊字符
|
||
"""
|
||
|
||
import logging
|
||
from cryptoai.utils.db_manager import get_db_manager
|
||
from sqlalchemy import text
|
||
|
||
# 配置日志
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
||
)
|
||
logger = logging.getLogger('update_db_charset')
|
||
|
||
def update_table_charset():
|
||
"""更新数据库表字符集为utf8mb4"""
|
||
try:
|
||
# 获取数据库管理器
|
||
db_manager = get_db_manager()
|
||
|
||
if not db_manager.engine:
|
||
logger.error("数据库连接失败")
|
||
return False
|
||
|
||
# 创建会话
|
||
session = db_manager.Session()
|
||
|
||
try:
|
||
# 更新数据库字符集
|
||
session.execute(text("ALTER DATABASE CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;"))
|
||
|
||
# 更新agent_feeds表字符集
|
||
session.execute(text("""
|
||
ALTER TABLE agent_feeds
|
||
CONVERT TO CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
|
||
"""))
|
||
|
||
# 特别更新content列的字符集
|
||
session.execute(text("""
|
||
ALTER TABLE agent_feeds
|
||
MODIFY content TEXT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
|
||
"""))
|
||
|
||
# 检查users表是否存在
|
||
result = session.execute(text("""
|
||
SELECT COUNT(*)
|
||
FROM information_schema.tables
|
||
WHERE table_schema = DATABASE()
|
||
AND table_name = 'users';
|
||
"""))
|
||
|
||
table_exists = result.scalar() > 0
|
||
|
||
# 如果users表存在，更新其字符集
|
||
if table_exists:
|
||
session.execute(text("""
|
||
ALTER TABLE users
|
||
CONVERT TO CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
|
||
"""))
|
||
|
||
# 特别更新nickname和mail列的字符集
|
||
session.execute(text("""
|
||
ALTER TABLE users
|
||
MODIFY nickname VARCHAR(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
|
||
"""))
|
||
|
||
session.execute(text("""
|
||
ALTER TABLE users
|
||
MODIFY mail VARCHAR(100) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
|
||
"""))
|
||
|
||
logger.info("成功更新users表字符集为utf8mb4")
|
||
|
||
# 检查user_questions表是否存在
|
||
result = session.execute(text("""
|
||
SELECT COUNT(*)
|
||
FROM information_schema.tables
|
||
WHERE table_schema = DATABASE()
|
||
AND table_name = 'user_questions';
|
||
"""))
|
||
|
||
table_exists = result.scalar() > 0
|
||
|
||
# 如果user_questions表存在，更新其字符集
|
||
if table_exists:
|
||
session.execute(text("""
|
||
ALTER TABLE user_questions
|
||
CONVERT TO CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
|
||
"""))
|
||
|
||
# 特别更新question列的字符集
|
||
session.execute(text("""
|
||
ALTER TABLE user_questions
|
||
MODIFY question TEXT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
|
||
"""))
|
||
|
||
logger.info("成功更新user_questions表字符集为utf8mb4")
|
||
|
||
session.commit()
|
||
logger.info("成功更新数据库表字符集为utf8mb4")
|
||
return True
|
||
|
||
except Exception as e:
|
||
session.rollback()
|
||
logger.error(f"更新数据库表字符集失败: {e}")
|
||
return False
|
||
|
||
finally:
|
||
session.close()
|
||
|
||
except Exception as e:
|
||
logger.error(f"更新数据库表字符集失败: {e}")
|
||
return False
|
||
|
||
if __name__ == "__main__":
|
||
update_table_charset() |