进行了 GVSDSDK 迁移，可能存在诸多问题
This commit is contained in:
532
gvsdsdk/msync/orm_handler.py
Normal file
532
gvsdsdk/msync/orm_handler.py
Normal file
@@ -0,0 +1,532 @@
|
||||
"""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
|
||||
Reference in New Issue
Block a user