533 lines
20 KiB
Python
533 lines
20 KiB
Python
"""MSYNC ORM 中转处理器
|
||
|
||
子服务器端注册 Socket.IO 事件处理器，接收主服务器发来的 ORM 查询请求，
|
||
在本地数据库执行 SQL 并返回结果。
|
||
|
||
用法（子服务器端）:
|
||
from gvsdsdk.msync.orm_handler import ORMRelayHandler
|
||
|
||
handler = ORMRelayHandler()
|
||
handler.Register(sio_client) # 注册到 MSyncSocketClient 实例
|
||
"""
|
||
|
||
import datetime
|
||
import logging
|
||
import uuid
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class ORMRelayHandler:
|
||
"""ORM 中转处理器——在子服务器端执行本地 SQL 查询
|
||
|
||
通过 MSYNC Socket.IO 事件协议，主服务器将 ORM 请求转发给子服务器，
|
||
子服务器在本地数据库执行查询后返回结果。
|
||
"""
|
||
|
||
def Register(self, sio_client):
|
||
"""将 ORM 事件处理器注册到 MSyncSocketClient 实例
|
||
|
||
Args:
|
||
sio_client: MSyncSocketClient 实例，需要有 _sio 属性
|
||
"""
|
||
sio = sio_client._sio
|
||
|
||
@sio.on('msync:orm:tables', namespace='/msync')
|
||
def on_orm_tables(data):
|
||
return self.HandleTables(data)
|
||
|
||
@sio.on('msync:orm:columns', namespace='/msync')
|
||
def on_orm_columns(data):
|
||
return self.HandleColumns(data)
|
||
|
||
@sio.on('msync:orm:list', namespace='/msync')
|
||
def on_orm_list(data):
|
||
return self.HandleList(data)
|
||
|
||
@sio.on('msync:orm:add', namespace='/msync')
|
||
def on_orm_add(data):
|
||
return self.HandleAdd(data)
|
||
|
||
@sio.on('msync:orm:modify', namespace='/msync')
|
||
def on_orm_modify(data):
|
||
return self.HandleModify(data)
|
||
|
||
@sio.on('msync:orm:delete', namespace='/msync')
|
||
def on_orm_delete(data):
|
||
return self.HandleDelete(data)
|
||
|
||
logger.info('[ORM] ORM 中转处理器已注册')
|
||
|
||
def _GetConnection(self):
|
||
"""获取本地数据库连接（Django default）"""
|
||
from django.db import connections
|
||
conn = connections['default']
|
||
conn.ensure_connection()
|
||
return conn
|
||
|
||
# ─── 事件处理 ──────────────────────────────────────────────
|
||
|
||
def HandleTables(self, data):
|
||
"""列出本地数据库中所有用户表"""
|
||
try:
|
||
conn = self._GetConnection()
|
||
except Exception as e:
|
||
return {'code': 1, 'msg': f'数据库连接失败: {e}'}
|
||
|
||
try:
|
||
with conn.cursor() as cursor:
|
||
tables = self._ListTables(cursor)
|
||
return {'code': 0, 'msg': 'success', 'data': tables}
|
||
except Exception as e:
|
||
logger.error(f'[ORM] HandleTables 异常: {e}')
|
||
return {'code': 1, 'msg': f'查询失败: {e}'}
|
||
|
||
def HandleColumns(self, data):
|
||
"""列出指定表的列信息"""
|
||
table_name = (data or {}).get('Table', '')
|
||
if not table_name:
|
||
return {'code': 1, 'msg': '缺少 Table 参数'}
|
||
|
||
try:
|
||
conn = self._GetConnection()
|
||
except Exception as e:
|
||
return {'code': 1, 'msg': f'数据库连接失败: {e}'}
|
||
|
||
try:
|
||
with conn.cursor() as cursor:
|
||
columns = self._ListColumns(cursor, table_name)
|
||
return {'code': 0, 'msg': 'success', 'data': columns}
|
||
except Exception as e:
|
||
logger.error(f'[ORM] HandleColumns 异常: {e}')
|
||
return {'code': 1, 'msg': f'查询失败: {e}'}
|
||
|
||
def HandleList(self, data):
|
||
"""分页查询表数据"""
|
||
table_name = (data or {}).get('Table', '')
|
||
if not table_name:
|
||
return {'code': 1, 'msg': '缺少 Table 参数'}
|
||
|
||
try:
|
||
conn = self._GetConnection()
|
||
except Exception as e:
|
||
return {'code': 1, 'msg': f'数据库连接失败: {e}'}
|
||
|
||
try:
|
||
with conn.cursor() as cursor:
|
||
result = self._QueryTable(cursor, table_name, data or {})
|
||
if result.get('error'):
|
||
return {'code': 1, 'msg': result['error'], 'data': None}
|
||
return {'code': 0, 'msg': 'success', 'data': result}
|
||
except Exception as e:
|
||
logger.error(f'[ORM] HandleList 异常: {e}')
|
||
return {'code': 1, 'msg': f'查询失败: {e}'}
|
||
|
||
def HandleAdd(self, data):
|
||
"""新增记录"""
|
||
table_name = (data or {}).get('Table', '')
|
||
record = (data or {}).get('Record', {})
|
||
if not table_name:
|
||
return {'code': 1, 'msg': '缺少 Table 参数'}
|
||
|
||
try:
|
||
conn = self._GetConnection()
|
||
except Exception as e:
|
||
return {'code': 1, 'msg': f'数据库连接失败: {e}'}
|
||
|
||
try:
|
||
with conn.cursor() as cursor:
|
||
result = self._AddRecord(cursor, table_name, record)
|
||
return result
|
||
except Exception as e:
|
||
logger.error(f'[ORM] HandleAdd 异常: {e}')
|
||
return {'code': 1, 'msg': f'新增失败: {e}'}
|
||
|
||
def HandleModify(self, data):
|
||
"""修改记录"""
|
||
table_name = (data or {}).get('Table', '')
|
||
record = (data or {}).get('Record', {})
|
||
if not table_name:
|
||
return {'code': 1, 'msg': '缺少 Table 参数'}
|
||
|
||
try:
|
||
conn = self._GetConnection()
|
||
except Exception as e:
|
||
return {'code': 1, 'msg': f'数据库连接失败: {e}'}
|
||
|
||
try:
|
||
with conn.cursor() as cursor:
|
||
result = self._ModifyRecord(cursor, table_name, record)
|
||
return result
|
||
except Exception as e:
|
||
logger.error(f'[ORM] HandleModify 异常: {e}')
|
||
return {'code': 1, 'msg': f'修改失败: {e}'}
|
||
|
||
def HandleDelete(self, data):
|
||
"""删除记录"""
|
||
table_name = (data or {}).get('Table', '')
|
||
record = (data or {}).get('Record', {})
|
||
if not table_name:
|
||
return {'code': 1, 'msg': '缺少 Table 参数'}
|
||
|
||
try:
|
||
conn = self._GetConnection()
|
||
except Exception as e:
|
||
return {'code': 1, 'msg': f'数据库连接失败: {e}'}
|
||
|
||
try:
|
||
with conn.cursor() as cursor:
|
||
result = self._DeleteRecord(cursor, table_name, record)
|
||
return result
|
||
except Exception as e:
|
||
logger.error(f'[ORM] HandleDelete 异常: {e}')
|
||
return {'code': 1, 'msg': f'删除失败: {e}'}
|
||
|
||
# ─── SQL 执行 ──────────────────────────────────────────────
|
||
|
||
def _ListTables(self, cursor):
|
||
"""列出所有用户表，含 i18n 标题"""
|
||
cursor.execute(
|
||
"SELECT TABLE_NAME, TABLE_ROWS, TABLE_COMMENT FROM information_schema.TABLES "
|
||
"WHERE TABLE_SCHEMA = DATABASE() AND TABLE_TYPE = 'BASE TABLE' "
|
||
"ORDER BY TABLE_NAME"
|
||
)
|
||
|
||
from django.apps import apps
|
||
# 构建表名→中文标题映射
|
||
model_verbose = {}
|
||
for model in apps.get_models():
|
||
meta = model._meta
|
||
# 优先使用 table_title（自定义 Meta 属性），其次 verbose_name
|
||
table_title = getattr(meta, 'table_title', None)
|
||
title = str(table_title) if table_title else (str(meta.verbose_name) if meta.verbose_name else '')
|
||
if title:
|
||
model_verbose[meta.object_name] = title
|
||
model_verbose[meta.db_table] = title
|
||
|
||
result = []
|
||
for row in cursor.fetchall():
|
||
table_name, table_rows, table_comment = row
|
||
# 优先级：TABLE_COMMENT > table_title/verbose_name > 表名
|
||
title = table_comment or model_verbose.get(table_name) or table_name
|
||
result.append({'name': table_name, 'title': title, 'count': table_rows or 0})
|
||
return result
|
||
|
||
def _ListColumns(self, cursor, table_name):
|
||
"""列出指定表的列信息"""
|
||
cursor.execute(
|
||
"SELECT COLUMN_NAME, DATA_TYPE, IS_NULLABLE, COLUMN_KEY, EXTRA, COLUMN_TYPE, COLUMN_COMMENT "
|
||
"FROM information_schema.COLUMNS "
|
||
"WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s "
|
||
"ORDER BY ORDINAL_POSITION",
|
||
[table_name]
|
||
)
|
||
|
||
model_verbose = self._FindModelVerboseNames(table_name)
|
||
logger.info(f'[ORM] _ListColumns table={table_name} model_verbose={model_verbose}')
|
||
|
||
columns = []
|
||
for row in cursor.fetchall():
|
||
col_name, data_type, is_nullable, col_key, extra, col_type, col_comment = row
|
||
is_pk = col_key == 'PRI'
|
||
is_auto = 'auto_increment' in (extra or '').lower()
|
||
|
||
if data_type in ('int', 'bigint', 'tinyint', 'smallint', 'mediumint'):
|
||
ftype = 'number'
|
||
elif data_type in ('float', 'double', 'decimal'):
|
||
ftype = 'number'
|
||
elif data_type in ('datetime', 'timestamp'):
|
||
ftype = 'datetime'
|
||
elif data_type in ('date',):
|
||
ftype = 'date'
|
||
elif data_type in ('time',):
|
||
ftype = 'time'
|
||
elif data_type in ('json',):
|
||
ftype = 'textarea'
|
||
elif data_type in ('text', 'mediumtext', 'longtext', 'tinytext'):
|
||
ftype = 'textarea'
|
||
elif data_type in ('binary', 'varbinary', 'blob'):
|
||
ftype = 'uuid'
|
||
else:
|
||
ftype = 'text'
|
||
|
||
title = col_comment or model_verbose.get(col_name) or col_name
|
||
|
||
columns.append({
|
||
'field': col_name,
|
||
'title': title,
|
||
'type': ftype,
|
||
'editable': not is_pk and not is_auto,
|
||
'sortable': True,
|
||
'filterable': True,
|
||
'primaryKey': is_pk,
|
||
'hideInForm': is_pk,
|
||
'required': is_nullable == 'NO' and not is_auto,
|
||
'dbType': col_type,
|
||
})
|
||
return columns
|
||
|
||
def _QueryTable(self, cursor, table_name, params):
|
||
"""分页查询表数据"""
|
||
cursor.execute(
|
||
"SELECT COUNT(*) FROM information_schema.TABLES "
|
||
"WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s",
|
||
[table_name]
|
||
)
|
||
if cursor.fetchone()[0] == 0:
|
||
return {'records': [], 'total': 0, 'error': f'表不存在: {table_name}'}
|
||
|
||
col_info = self._GetTableColumns(cursor, table_name)
|
||
pk_cols = {k: v for k, v in col_info.items() if v['is_pk']}
|
||
|
||
where_parts = []
|
||
where_values = []
|
||
for pk_name in pk_cols:
|
||
pk_val = params.get(pk_name, '')
|
||
if pk_val:
|
||
where_parts.append(f'`{pk_name}` = %s')
|
||
where_values.append(self._ConvertValueForDB(pk_val, pk_cols[pk_name]))
|
||
|
||
keyword = params.get('keyword', '')
|
||
if not where_parts and keyword:
|
||
cursor.execute(
|
||
"SELECT COLUMN_NAME FROM information_schema.COLUMNS "
|
||
"WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s "
|
||
"AND DATA_TYPE IN ('varchar', 'char', 'text', 'mediumtext', 'longtext')",
|
||
[table_name]
|
||
)
|
||
text_cols = [row[0] for row in cursor.fetchall()]
|
||
if text_cols:
|
||
conditions = [f'`{c}` LIKE %s' for c in text_cols]
|
||
or_clause = ' OR '.join(conditions)
|
||
where_parts.append(f'({or_clause})')
|
||
where_values.extend([f'%{keyword}%'] * len(text_cols))
|
||
|
||
where_clause = (' WHERE ' + ' AND '.join(where_parts)) if where_parts else ''
|
||
|
||
if where_clause:
|
||
cursor.execute(f'SELECT COUNT(*) FROM `{table_name}`{where_clause}', where_values)
|
||
total = cursor.fetchone()[0]
|
||
else:
|
||
cursor.execute(f'SELECT COUNT(*) FROM `{table_name}`')
|
||
total = cursor.fetchone()[0]
|
||
|
||
try:
|
||
page = max(int(params.get('PageIndex', 1)), 1)
|
||
except (ValueError, TypeError):
|
||
page = 1
|
||
try:
|
||
page_size = min(max(int(params.get('PageSize', 20)), 1), 200)
|
||
except (ValueError, TypeError):
|
||
page_size = 20
|
||
|
||
offset = (page - 1) * page_size
|
||
cursor.execute(
|
||
f'SELECT * FROM `{table_name}`{where_clause} ORDER BY 1 LIMIT %s OFFSET %s',
|
||
where_values + [page_size, offset]
|
||
)
|
||
desc = [d[0] for d in cursor.description]
|
||
records = []
|
||
for row in cursor.fetchall():
|
||
record = {}
|
||
for i, val in enumerate(row):
|
||
val = self._SerializeValue(val)
|
||
record[desc[i]] = val
|
||
records.append(record)
|
||
|
||
return {'records': records, 'total': total, 'page': page, 'PageSize': page_size}
|
||
|
||
def _AddRecord(self, cursor, table_name, record):
|
||
"""新增记录"""
|
||
if not self._ValidateTableName(cursor, table_name):
|
||
return {'code': 1, 'msg': f'表 {table_name} 不存在'}
|
||
|
||
col_info = self._GetTableColumns(cursor, table_name)
|
||
|
||
fields = []
|
||
values = []
|
||
for key, val in record.items():
|
||
if key not in col_info:
|
||
continue
|
||
if col_info[key]['is_auto']:
|
||
continue
|
||
fields.append(f'`{key}`')
|
||
values.append(self._ConvertValueForDB(val, col_info[key]))
|
||
|
||
if not fields:
|
||
return {'code': 1, 'msg': '没有可插入的字段'}
|
||
|
||
placeholders = ', '.join(['%s'] * len(fields))
|
||
sql = f'INSERT INTO `{table_name}` ({", ".join(fields)}) VALUES ({placeholders})'
|
||
cursor.execute(sql, values)
|
||
return {'code': 0, 'msg': '新增成功'}
|
||
|
||
def _ModifyRecord(self, cursor, table_name, record):
|
||
"""修改记录"""
|
||
if not self._ValidateTableName(cursor, table_name):
|
||
return {'code': 1, 'msg': f'表 {table_name} 不存在'}
|
||
|
||
col_info = self._GetTableColumns(cursor, table_name)
|
||
pk_cols = {k: v for k, v in col_info.items() if v['is_pk']}
|
||
|
||
if not pk_cols:
|
||
return {'code': 1, 'msg': '表没有主键，无法定位记录'}
|
||
|
||
set_parts = []
|
||
set_values = []
|
||
for key, val in record.items():
|
||
if key not in col_info or key in pk_cols:
|
||
continue
|
||
set_parts.append(f'`{key}` = %s')
|
||
set_values.append(self._ConvertValueForDB(val, col_info[key]))
|
||
|
||
if not set_parts:
|
||
return {'code': 0, 'msg': '没有需要更新的字段'}
|
||
|
||
where_parts = []
|
||
where_values = []
|
||
for pk_name in pk_cols:
|
||
if pk_name not in record:
|
||
return {'code': 1, 'msg': f'缺少主键字段: {pk_name}'}
|
||
where_parts.append(f'`{pk_name}` = %s')
|
||
where_values.append(self._ConvertValueForDB(record[pk_name], pk_cols[pk_name]))
|
||
|
||
sql = f'UPDATE `{table_name}` SET {", ".join(set_parts)} WHERE {" AND ".join(where_parts)}'
|
||
cursor.execute(sql, set_values + where_values)
|
||
|
||
if cursor.rowcount == 0:
|
||
return {'code': 1, 'msg': '记录不存在或未变更'}
|
||
return {'code': 0, 'msg': '修改成功'}
|
||
|
||
def _DeleteRecord(self, cursor, table_name, record):
|
||
"""删除记录"""
|
||
if not self._ValidateTableName(cursor, table_name):
|
||
return {'code': 1, 'msg': f'表 {table_name} 不存在'}
|
||
|
||
col_info = self._GetTableColumns(cursor, table_name)
|
||
pk_cols = {k: v for k, v in col_info.items() if v['is_pk']}
|
||
|
||
if not pk_cols:
|
||
return {'code': 1, 'msg': '表没有主键，无法定位记录'}
|
||
|
||
where_parts = []
|
||
where_values = []
|
||
for pk_name in pk_cols:
|
||
if pk_name not in record:
|
||
return {'code': 1, 'msg': f'缺少主键字段: {pk_name}'}
|
||
where_parts.append(f'`{pk_name}` = %s')
|
||
where_values.append(self._ConvertValueForDB(record[pk_name], pk_cols[pk_name]))
|
||
|
||
sql = f'DELETE FROM `{table_name}` WHERE {" AND ".join(where_parts)}'
|
||
cursor.execute(sql, where_values)
|
||
|
||
if cursor.rowcount == 0:
|
||
return {'code': 1, 'msg': '记录不存在'}
|
||
return {'code': 0, 'msg': '删除成功'}
|
||
|
||
# ─── 工具方法 ──────────────────────────────────────────────
|
||
|
||
def _ValidateTableName(self, cursor, table_name):
|
||
"""验证表名存在"""
|
||
cursor.execute(
|
||
"SELECT COUNT(*) FROM information_schema.TABLES "
|
||
"WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s",
|
||
[table_name]
|
||
)
|
||
return cursor.fetchone()[0] > 0
|
||
|
||
def _GetTableColumns(self, cursor, table_name):
|
||
"""获取表的列信息"""
|
||
cursor.execute(
|
||
"SELECT COLUMN_NAME, DATA_TYPE, COLUMN_TYPE, COLUMN_KEY, EXTRA, IS_NULLABLE "
|
||
"FROM information_schema.COLUMNS "
|
||
"WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s "
|
||
"ORDER BY ORDINAL_POSITION",
|
||
[table_name]
|
||
)
|
||
cols = {}
|
||
for row in cursor.fetchall():
|
||
col_name, data_type, column_type, col_key, extra, is_nullable = row
|
||
cols[col_name] = {
|
||
'type': data_type,
|
||
'column_type': column_type or '',
|
||
'is_pk': col_key == 'PRI',
|
||
'is_auto': 'auto_increment' in (extra or '').lower(),
|
||
'is_nullable': is_nullable == 'YES',
|
||
}
|
||
return cols
|
||
|
||
def _ConvertValueForDB(self, value, col_info):
|
||
"""将前端传来的值转换为数据库可接受的类型"""
|
||
if value is None or value == '':
|
||
if col_info['is_nullable']:
|
||
return None
|
||
return ''
|
||
|
||
data_type = col_info['type']
|
||
|
||
if data_type in ('int', 'bigint', 'tinyint', 'smallint', 'mediumint'):
|
||
return int(value)
|
||
elif data_type in ('float', 'double', 'decimal'):
|
||
return float(value)
|
||
elif data_type in ('binary', 'varbinary'):
|
||
import re
|
||
hex_str = re.sub(r'[^0-9a-fA-F]', '', str(value))
|
||
if len(hex_str) == 32:
|
||
return uuid.UUID(hex_str).bytes
|
||
return value
|
||
|
||
return value
|
||
|
||
def _SerializeValue(self, val):
|
||
"""将数据库值序列化为 JSON 安全格式"""
|
||
if isinstance(val, bytes):
|
||
if len(val) == 16:
|
||
try:
|
||
return str(uuid.UUID(bytes=val))
|
||
except ValueError:
|
||
return val.hex()
|
||
return val.hex()
|
||
elif isinstance(val, datetime.datetime):
|
||
return val.strftime('%Y-%m-%d %H:%M:%S')
|
||
elif isinstance(val, datetime.date):
|
||
return val.isoformat()
|
||
elif isinstance(val, datetime.time):
|
||
return val.strftime('%H:%M:%S')
|
||
elif isinstance(val, (bytearray, memoryview)):
|
||
return bytes(val).hex()
|
||
return val
|
||
|
||
def _FindModelVerboseNames(self, table_name):
|
||
"""从 Django 模型中获取字段的 verbose_name"""
|
||
from django.apps import apps
|
||
verbose_map = {}
|
||
try:
|
||
for model in apps.get_models():
|
||
if model._meta.db_table == table_name:
|
||
logger.info(f'[ORM] _FindModelVerboseNames matched model={model.__name__} db_table={model._meta.db_table}')
|
||
for f in model._meta.get_fields():
|
||
# 获取数据库列名
|
||
col_name = None
|
||
if hasattr(f, 'column') and f.column:
|
||
col_name = f.column
|
||
elif hasattr(f, 'db_column') and f.db_column:
|
||
col_name = f.db_column
|
||
elif hasattr(f, 'attname'):
|
||
col_name = f.attname
|
||
|
||
if not col_name:
|
||
continue
|
||
|
||
vn = getattr(f, 'verbose_name', None)
|
||
if vn and str(vn) != col_name:
|
||
verbose_map[col_name] = str(vn)
|
||
logger.debug(f'[ORM] {col_name} -> {str(vn)}')
|
||
break
|
||
else:
|
||
# 没有匹配的模型
|
||
available = [m._meta.db_table for m in apps.get_models()]
|
||
logger.warning(f'[ORM] _FindModelVerboseNames: 未找到 db_table={table_name} 的模型, 可用表: {available[:20]}...')
|
||
except Exception as e:
|
||
logger.error(f'[ORM] _FindModelVerboseNames 异常: {e}')
|
||
return verbose_map
|