288 lines
11 KiB
Python
288 lines
11 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
数据库连接配置
|
||
"""
|
||
|
||
import pymysql
|
||
import logging
|
||
import time
|
||
from datetime import datetime
|
||
from pymysql.cursors import DictCursor
|
||
from contextlib import contextmanager
|
||
from pymysql.err import OperationalError, InterfaceError, InternalError
|
||
import threading
|
||
|
||
# 数据库配置
|
||
DB_CONFIG = {
|
||
'host': '8.149.233.36',
|
||
'port': 3306,
|
||
'user': 'ai_wht_write',
|
||
'password': '7aK_H2yvokVumr84lLNDt8fDBp6P',
|
||
'database': 'ai_wht',
|
||
'charset': 'utf8mb4',
|
||
'collation': 'utf8mb4_unicode_ci',
|
||
'cursorclass': DictCursor,
|
||
'autocommit': True,
|
||
'connect_timeout': 10,
|
||
'read_timeout': 30,
|
||
'write_timeout': 30,
|
||
'max_allowed_packet': 16777216, # 16MB
|
||
'sql_mode': 'STRICT_TRANS_TABLES,NO_ZERO_DATE,NO_ZERO_IN_DATE,ERROR_FOR_DIVISION_BY_ZERO'
|
||
}
|
||
|
||
# 使用与主应用相同的日志记录器
|
||
logger = logging.getLogger('article_server')
|
||
|
||
class DatabaseManager:
|
||
"""数据库连接管理器"""
|
||
|
||
def __init__(self, config=None):
|
||
self.config = config or DB_CONFIG
|
||
# 使用线程本地连接，避免多线程复用同一连接导致的数据包序号错误
|
||
self._local = threading.local()
|
||
self._max_retries = 3
|
||
self._retry_delay = 1 # 秒
|
||
|
||
def get_connection(self):
|
||
"""获取数据库连接"""
|
||
try:
|
||
# 线程本地连接对象
|
||
connection = getattr(self._local, 'connection', None)
|
||
|
||
# 检查现有连接是否有效
|
||
if connection is not None:
|
||
try:
|
||
# 尝试 ping 并在需要时自动重连
|
||
connection.ping(reconnect=True)
|
||
if connection.open:
|
||
return connection
|
||
except (OperationalError, InterfaceError, AttributeError):
|
||
logger.warning("检测到无效连接，将重新创建")
|
||
connection = None
|
||
self._local.connection = None
|
||
|
||
# 创建新连接（线程专有）
|
||
self._local.connection = pymysql.connect(**self.config)
|
||
|
||
# 确保连接使用正确的字符集
|
||
with self._local.connection.cursor() as cursor:
|
||
cursor.execute("SET NAMES utf8mb4")
|
||
|
||
logger.info("数据库连接成功，字符集设置完成")
|
||
return self._local.connection
|
||
|
||
except Exception as e:
|
||
logger.error(f"数据库连接失败: {e}")
|
||
# 确保线程本地连接被清理
|
||
try:
|
||
self._local.connection = None
|
||
except Exception:
|
||
pass
|
||
raise
|
||
|
||
def close_connection(self):
|
||
"""关闭数据库连接"""
|
||
connection = getattr(self._local, 'connection', None)
|
||
if connection:
|
||
try:
|
||
if connection.open:
|
||
connection.close()
|
||
logger.info("数据库连接已关闭")
|
||
except Exception as e:
|
||
logger.warning(f"关闭数据库连接时出错: {e}")
|
||
finally:
|
||
self._local.connection = None
|
||
|
||
def _execute_with_retry(self, operation, *args, **kwargs):
|
||
"""带重试机制的执行操作"""
|
||
last_exception = None
|
||
|
||
for attempt in range(self._max_retries):
|
||
try:
|
||
return operation(*args, **kwargs)
|
||
except (OperationalError, InterfaceError, InternalError, AttributeError) as e:
|
||
last_exception = e
|
||
error_code = getattr(e, 'args', [None])[0] if getattr(e, 'args', None) else None
|
||
|
||
# 检查是否是连接丢失错误
|
||
message = str(e)
|
||
is_packet_seq_error = isinstance(e, InternalError) and 'Packet sequence number wrong' in message
|
||
is_already_closed = isinstance(e, InterfaceError) and ('Already closed' in message or 'closed' in message.lower())
|
||
is_none_read = isinstance(e, AttributeError) and "'NoneType' object has no attribute 'read'" in message
|
||
if error_code in [2006, 2013, 2055] or is_packet_seq_error or is_already_closed or is_none_read:
|
||
logger.warning(f"数据库连接丢失 (尝试 {attempt + 1}/{self._max_retries}): {e}")
|
||
# 关闭现有连接
|
||
self.close_connection()
|
||
|
||
if attempt < self._max_retries - 1:
|
||
time.sleep(self._retry_delay)
|
||
continue
|
||
else:
|
||
# 其他数据库错误，不重试
|
||
raise
|
||
except Exception as e:
|
||
# 其他错误，不重试
|
||
raise
|
||
|
||
# 所有重试都失败了
|
||
logger.error(f"数据库操作失败，已重试 {self._max_retries} 次: {last_exception}")
|
||
raise last_exception
|
||
|
||
@contextmanager
|
||
def get_cursor(self):
|
||
"""获取数据库游标的上下文管理器"""
|
||
connection = None
|
||
cursor = None
|
||
|
||
try:
|
||
# 获取连接
|
||
connection = self._execute_with_retry(self.get_connection)
|
||
cursor = connection.cursor(DictCursor)
|
||
yield cursor
|
||
|
||
# 提交事务
|
||
if connection and connection.open:
|
||
try:
|
||
connection.commit()
|
||
except (OperationalError, InterfaceError) as e:
|
||
logger.warning(f"提交事务失败: {e}")
|
||
# 不抛出异常，因为可能是连接问题
|
||
|
||
except Exception as e:
|
||
# 回滚事务
|
||
if connection and connection.open:
|
||
try:
|
||
connection.rollback()
|
||
except (OperationalError, InterfaceError) as rollback_error:
|
||
logger.warning(f"回滚事务失败: {rollback_error}")
|
||
# 不抛出异常，因为可能是连接问题
|
||
|
||
logger.error(f"数据库操作失败: {e}")
|
||
raise
|
||
finally:
|
||
# 关闭游标
|
||
if cursor:
|
||
try:
|
||
cursor.close()
|
||
except Exception as e:
|
||
logger.warning(f"关闭游标时出错: {e}")
|
||
|
||
def execute_query(self, sql, params=None):
|
||
"""执行查询SQL"""
|
||
def _query_operation():
|
||
try:
|
||
with self.get_cursor() as cursor:
|
||
# 记录SQL执行开始
|
||
logger.info(f"[SQL执行] 开始执行查询SQL")
|
||
logger.info(f"[SQL语句] {sql}")
|
||
if params:
|
||
logger.info(f"[SQL参数] {params}")
|
||
|
||
cursor.execute(sql, params)
|
||
result = cursor.fetchall()
|
||
|
||
# 记录SQL执行结果
|
||
logger.info(f"[SQL结果] 查询完成，返回 {len(result)} 条记录")
|
||
if result and len(result) <= 3: # 只记录前3条结果，避免日志过长
|
||
logger.info(f"[SQL数据] 查询结果: {result}")
|
||
|
||
return result
|
||
except Exception as e:
|
||
# 异常记录到错误日志
|
||
logger.error(f"[SQL执行失败] 查询SQL执行异常: {e}")
|
||
logger.error(f"[SQL语句] {sql}")
|
||
if params:
|
||
logger.error(f"[SQL参数] {params}")
|
||
raise
|
||
|
||
return self._execute_with_retry(_query_operation)
|
||
|
||
def execute_update(self, sql, params=None):
|
||
"""执行更新SQL"""
|
||
def _update_operation():
|
||
try:
|
||
with self.get_cursor() as cursor:
|
||
# 记录SQL执行开始
|
||
logger.info(f"[SQL执行] 开始执行更新SQL")
|
||
logger.info(f"[SQL语句] {sql}")
|
||
if params:
|
||
logger.info(f"[SQL参数] {params}")
|
||
|
||
affected_rows = cursor.execute(sql, params)
|
||
|
||
# 记录SQL执行结果
|
||
logger.info(f"[SQL结果] 更新完成，影响 {affected_rows} 行")
|
||
|
||
return affected_rows
|
||
except Exception as e:
|
||
# 异常记录到错误日志
|
||
logger.error(f"[SQL执行失败] 更新SQL执行异常: {e}")
|
||
logger.error(f"[SQL语句] {sql}")
|
||
if params:
|
||
logger.error(f"[SQL参数] {params}")
|
||
raise
|
||
|
||
return self._execute_with_retry(_update_operation)
|
||
|
||
def execute_insert(self, sql, params=None):
|
||
"""执行插入SQL"""
|
||
def _insert_operation():
|
||
try:
|
||
with self.get_cursor() as cursor:
|
||
# 记录SQL执行开始
|
||
logger.info(f"[SQL执行] 开始执行插入SQL")
|
||
logger.info(f"[SQL语句] {sql}")
|
||
if params:
|
||
logger.info(f"[SQL参数] {params}")
|
||
|
||
cursor.execute(sql, params)
|
||
last_id = cursor.lastrowid
|
||
|
||
# 记录SQL执行结果
|
||
logger.info(f"[SQL结果] 插入完成，新记录ID: {last_id}")
|
||
|
||
return last_id
|
||
except Exception as e:
|
||
# 异常记录到错误日志
|
||
logger.error(f"[SQL执行失败] 插入SQL执行异常: {e}")
|
||
logger.error(f"[SQL语句] {sql}")
|
||
if params:
|
||
logger.error(f"[SQL参数] {params}")
|
||
raise
|
||
|
||
return self._execute_with_retry(_insert_operation)
|
||
|
||
# 全局数据库管理器实例
|
||
db_manager = DatabaseManager()
|
||
|
||
def get_db_manager():
|
||
"""获取数据库管理器实例"""
|
||
return db_manager
|
||
|
||
def format_datetime_fields(data, datetime_fields=['created_at', 'updated_at']):
|
||
"""
|
||
格式化数据中的日期时间字段为 Y-m-d H:i:s 格式
|
||
|
||
Args:
|
||
data: 单个字典或字典列表
|
||
datetime_fields: 需要格式化的字段名列表
|
||
|
||
Returns:
|
||
格式化后的数据
|
||
"""
|
||
def format_single_record(record):
|
||
if not isinstance(record, dict):
|
||
return record
|
||
|
||
for field in datetime_fields:
|
||
if field in record and record[field] is not None:
|
||
if isinstance(record[field], datetime):
|
||
record[field] = record[field].strftime('%Y-%m-%d %H:%M:%S')
|
||
return record
|
||
|
||
if isinstance(data, list):
|
||
return [format_single_record(record) for record in data]
|
||
else:
|
||
return format_single_record(data)
|