mirror of
https://gh-proxy.org/https://github.com/klinsmaya/SQLBot.git
synced 2026-11-08 09:17:06 +08:00
293 lines
12 KiB
Python
293 lines
12 KiB
Python
# Author: Junjun
|
|
# Date: 2025/6/25
|
|
|
|
from typing import List, Dict
|
|
|
|
from apps.datasource.models.datasource import CoreField, CoreDatasource
|
|
from apps.db.constant import DB
|
|
from apps.system.models.system_variable_model import SystemVariable
|
|
from common.core.deps import SessionDep, CurrentUser
|
|
|
|
|
|
def _escape_sql_value(value: str) -> str:
|
|
"""Escape a string value for safe inclusion in a SQL literal.
|
|
|
|
Replaces single quotes with two single quotes (standard SQL escaping)
|
|
and strips characters that could break out of the string context.
|
|
"""
|
|
if value is None:
|
|
return value
|
|
# Standard SQL escaping: double any embedded single-quote characters
|
|
escaped = str(value).replace("'", "''")
|
|
# Remove backslashes that some drivers interpret as escape characters
|
|
escaped = escaped.replace("\\", "\\\\")
|
|
return escaped
|
|
|
|
|
|
def transFilterTree(session: SessionDep, current_user: CurrentUser, tree_list: List[any],
|
|
ds: CoreDatasource) -> str | None:
|
|
if tree_list is None:
|
|
return None
|
|
res: List[str] = []
|
|
for dto in tree_list:
|
|
tree = dto.tree
|
|
if tree is None:
|
|
continue
|
|
tree_exp = transTreeToWhere(session, current_user, tree, ds)
|
|
if tree_exp is not None:
|
|
res.append(tree_exp)
|
|
return " AND ".join(res)
|
|
|
|
|
|
_VALID_LOGIC_OPS = {"AND", "OR"}
|
|
|
|
|
|
def transTreeToWhere(session: SessionDep, current_user: CurrentUser, tree: any, ds: CoreDatasource) -> str | None:
|
|
if tree is None:
|
|
return None
|
|
logic = tree['logic']
|
|
# Validate the logic operator to prevent injection via this field
|
|
if logic.upper() not in _VALID_LOGIC_OPS:
|
|
return None
|
|
|
|
items = tree['items']
|
|
list: List[str] = []
|
|
if items is not None:
|
|
for item in items:
|
|
exp: str = None
|
|
if item['type'] == 'item':
|
|
exp = transTreeItem(session, current_user, item, ds)
|
|
elif item['type'] == 'tree':
|
|
exp = transTreeToWhere(session, current_user, item['sub_tree'], ds)
|
|
|
|
if exp is not None:
|
|
list.append(exp)
|
|
return '(' + f' {logic} '.join(list) + ')' if len(list) > 0 else None
|
|
|
|
|
|
def transTreeItem(session: SessionDep, current_user: CurrentUser, item: Dict, ds: CoreDatasource) -> str | None:
|
|
res: str = None
|
|
field = session.query(CoreField).filter(CoreField.id == int(item['field_id'])).first()
|
|
if field is None:
|
|
return None
|
|
|
|
db = DB.get_db(ds.type)
|
|
whereName = db.prefix + field.field_name + db.suffix
|
|
whereTerm = transFilterTerm(item['term'])
|
|
|
|
if item['filter_type'] == 'enum':
|
|
if len(item['enum_value']) > 0:
|
|
escaped_values = [_escape_sql_value(v) for v in item['enum_value']]
|
|
if ds['type'] == 'sqlServer' and (
|
|
field.field_type == 'nchar' or field.field_type == 'NCHAR' or field.field_type == 'nvarchar' or field.field_type == 'NVARCHAR'):
|
|
res = "(" + whereName + " IN (N'" + "',N'".join(escaped_values) + "'))"
|
|
else:
|
|
res = "(" + whereName + " IN ('" + "','".join(escaped_values) + "'))"
|
|
else:
|
|
# if system variable, do check and get value
|
|
# new field: value_type(variable or normal), variable_id
|
|
value_type = item.get('value_type')
|
|
if value_type and value_type == 'variable':
|
|
# get system variable
|
|
variable_id = item.get('variable_id')
|
|
if variable_id is not None:
|
|
sys_variable = session.query(SystemVariable).filter(SystemVariable.id == variable_id).first()
|
|
if sys_variable is None:
|
|
return None
|
|
|
|
# do inner system variable
|
|
if sys_variable.type == 'system':
|
|
res = whereName + whereTerm + getSysVariableValue(sys_variable, current_user, ds, field, item)
|
|
else:
|
|
# check user variable
|
|
user_variables = current_user.system_variables
|
|
if user_variables is None or len(user_variables) == 0 or not userHaveVariable(user_variables,
|
|
sys_variable):
|
|
return None
|
|
else:
|
|
# get user variable
|
|
u_variable = None
|
|
for u in user_variables:
|
|
if u.get('variableId') == sys_variable.id:
|
|
u_variable = u
|
|
break
|
|
if u_variable is None:
|
|
return None
|
|
|
|
# check value
|
|
values = u_variable.get('variableValues')
|
|
if sys_variable.var_type == 'text':
|
|
set_sys = set(sys_variable.value)
|
|
values = [x for x in values if x in set_sys]
|
|
if values is None or len(values) == 0:
|
|
return None
|
|
elif sys_variable.var_type == 'number':
|
|
if (sys_variable.value[0] is not None and values[0] < sys_variable.value[0]) or (
|
|
sys_variable.value[1] is not None and values[0] > sys_variable.value[1]):
|
|
return None
|
|
elif sys_variable.var_type == 'datetime':
|
|
if (sys_variable.value[0] is not None and values[0] < sys_variable.value[0]) or (
|
|
sys_variable.value[1] is not None and values[0] > sys_variable.value[1]):
|
|
return None
|
|
elif sys_variable.var_type == 'kv':
|
|
key_set = set(values)
|
|
values = [item["value"] for item in sys_variable.value if item["value"] in key_set]
|
|
if values is None or len(values) == 0:
|
|
return None
|
|
|
|
# build exp
|
|
whereValue = ''
|
|
if item['term'] == 'null':
|
|
whereValue = ''
|
|
elif item['term'] == 'not_null':
|
|
whereValue = ''
|
|
elif item['term'] == 'empty':
|
|
whereValue = "''"
|
|
elif item['term'] == 'not_empty':
|
|
whereValue = "''"
|
|
elif item['term'] == 'in' or item['term'] == 'not in':
|
|
escaped_values = [_escape_sql_value(v) for v in values]
|
|
if ds.type == 'sqlServer' and (
|
|
field.field_type == 'nchar' or field.field_type == 'NCHAR' or field.field_type == 'nvarchar' or field.field_type == 'NVARCHAR'):
|
|
whereValue = "(N'" + "', N'".join(escaped_values) + "')"
|
|
else:
|
|
whereValue = "('" + "', '".join(escaped_values) + "')"
|
|
elif item['term'] == 'like' or item['term'] == 'not like':
|
|
escaped_v = _escape_sql_value(values[0])
|
|
if ds.type == 'sqlServer' and (
|
|
field.field_type == 'nchar' or field.field_type == 'NCHAR' or field.field_type == 'nvarchar' or field.field_type == 'NVARCHAR'):
|
|
whereValue = f"N'%{escaped_v}%'"
|
|
else:
|
|
whereValue = f"'%{escaped_v}%'"
|
|
else:
|
|
escaped_v = _escape_sql_value(values[0])
|
|
if ds.type == 'sqlServer' and (
|
|
field.field_type == 'nchar' or field.field_type == 'NCHAR' or field.field_type == 'nvarchar' or field.field_type == 'NVARCHAR'):
|
|
whereValue = f"N'{escaped_v}'"
|
|
else:
|
|
whereValue = f"'{escaped_v}'"
|
|
|
|
res = whereName + whereTerm + whereValue
|
|
else:
|
|
return None
|
|
else:
|
|
value = item['value']
|
|
whereValue = ''
|
|
|
|
if item['term'] == 'null':
|
|
whereValue = ''
|
|
elif item['term'] == 'not_null':
|
|
whereValue = ''
|
|
elif item['term'] == 'empty':
|
|
whereValue = "''"
|
|
elif item['term'] == 'not_empty':
|
|
whereValue = "''"
|
|
elif item['term'] == 'in' or item['term'] == 'not in':
|
|
escaped_values = [_escape_sql_value(v) for v in value.split(",")]
|
|
if ds.type == 'sqlServer' and (
|
|
field.field_type == 'nchar' or field.field_type == 'NCHAR' or field.field_type == 'nvarchar' or field.field_type == 'NVARCHAR'):
|
|
whereValue = "(N'" + "', N'".join(escaped_values) + "')"
|
|
else:
|
|
whereValue = "('" + "', '".join(escaped_values) + "')"
|
|
elif item['term'] == 'like' or item['term'] == 'not like':
|
|
escaped_v = _escape_sql_value(value)
|
|
if ds.type == 'sqlServer' and (
|
|
field.field_type == 'nchar' or field.field_type == 'NCHAR' or field.field_type == 'nvarchar' or field.field_type == 'NVARCHAR'):
|
|
whereValue = f"N'%{escaped_v}%'"
|
|
else:
|
|
whereValue = f"'%{escaped_v}%'"
|
|
else:
|
|
escaped_v = _escape_sql_value(value)
|
|
if ds.type == 'sqlServer' and (
|
|
field.field_type == 'nchar' or field.field_type == 'NCHAR' or field.field_type == 'nvarchar' or field.field_type == 'NVARCHAR'):
|
|
whereValue = f"N'{escaped_v}'"
|
|
else:
|
|
whereValue = f"'{escaped_v}'"
|
|
|
|
res = whereName + whereTerm + whereValue
|
|
return res
|
|
|
|
|
|
def transFilterTerm(term: str) -> str:
|
|
if term == "eq":
|
|
return " = "
|
|
if term == "not_eq":
|
|
return " <> "
|
|
if term == "lt":
|
|
return " < "
|
|
if term == "le":
|
|
return " <= "
|
|
if term == "gt":
|
|
return " > "
|
|
if term == "ge":
|
|
return " >= "
|
|
if term == "in":
|
|
return " IN "
|
|
if term == "not in":
|
|
return " NOT IN "
|
|
if term == "like":
|
|
return " LIKE "
|
|
if term == "not like":
|
|
return " NOT LIKE "
|
|
if term == "null":
|
|
return " IS NULL "
|
|
if term == "not_null":
|
|
return " IS NOT NULL "
|
|
if term == "empty":
|
|
return " = "
|
|
if term == "not_empty":
|
|
return " <> "
|
|
if term == "between":
|
|
return " BETWEEN "
|
|
return ""
|
|
|
|
|
|
def userHaveVariable(user_variables: List, sys_variable: SystemVariable):
|
|
for u in user_variables:
|
|
if sys_variable.id == u.get('variableId'):
|
|
return True
|
|
return False
|
|
|
|
|
|
def getSysVariableValue(sys_variable: SystemVariable, current_user: CurrentUser, ds: CoreDatasource, field: CoreField,
|
|
item: Dict, ):
|
|
v = None
|
|
if sys_variable.value[0] == 'name':
|
|
v = current_user.name
|
|
if sys_variable.value[0] == 'account':
|
|
v = current_user.account
|
|
if sys_variable.value[0] == 'email':
|
|
v = current_user.email
|
|
|
|
escaped_v = _escape_sql_value(v) if v is not None else v
|
|
|
|
whereValue = ''
|
|
if item['term'] == 'null':
|
|
whereValue = ''
|
|
elif item['term'] == 'not_null':
|
|
whereValue = ''
|
|
elif item['term'] == 'empty':
|
|
whereValue = "''"
|
|
elif item['term'] == 'not_empty':
|
|
whereValue = "''"
|
|
elif item['term'] == 'in' or item['term'] == 'not in':
|
|
if ds.type == 'sqlServer' and (
|
|
field.field_type == 'nchar' or field.field_type == 'NCHAR' or field.field_type == 'nvarchar' or field.field_type == 'NVARCHAR'):
|
|
whereValue = f"(N'{escaped_v}')"
|
|
else:
|
|
whereValue = f"('{escaped_v}')"
|
|
elif item['term'] == 'like' or item['term'] == 'not like':
|
|
if ds.type == 'sqlServer' and (
|
|
field.field_type == 'nchar' or field.field_type == 'NCHAR' or field.field_type == 'nvarchar' or field.field_type == 'NVARCHAR'):
|
|
whereValue = f"N'%{escaped_v}%'"
|
|
else:
|
|
whereValue = f"'%{escaped_v}%'"
|
|
else:
|
|
if ds.type == 'sqlServer' and (
|
|
field.field_type == 'nchar' or field.field_type == 'NCHAR' or field.field_type == 'nvarchar' or field.field_type == 'NVARCHAR'):
|
|
whereValue = f"N'{escaped_v}'"
|
|
else:
|
|
whereValue = f"'{escaped_v}'"
|
|
|
|
return whereValue
|