120 lines
7.1 KiB
Python
120 lines
7.1 KiB
Python
import datetime
|
|
import random
|
|
import time
|
|
from typing import List
|
|
|
|
from sqlgenerator.ColumnValue import ValueType, ColumnValue
|
|
from sqlparse.SQLColumn import SQLColumn, SQLColumnType
|
|
from sqlparse.SQLTable import SQLTable
|
|
from lorem_text import lorem
|
|
|
|
min_bigint = 0
|
|
max_bigint = 9223372036854775807
|
|
|
|
min_int = -2147483648
|
|
max_int = 2147483647
|
|
|
|
min_date = datetime.date(2023, 1, 1)
|
|
max_date = datetime.date(2024, 12, 31)
|
|
|
|
min_time = datetime.time(0, 0, 0)
|
|
max_time = datetime.time(23, 59, 59)
|
|
|
|
|
|
def generate_random_insert(sql_table: SQLTable):
|
|
columnValues = []
|
|
for column in sql_table.columns:
|
|
columnValues.append(generate_random_column(column, columnValues, True))
|
|
|
|
sql_insert_command = "INSERT INTO " + sql_table.table_name + " VALUES (${i});"
|
|
for columnValue in columnValues:
|
|
if columnValue.valueType == ValueType.INT:
|
|
sql_insert_command = sql_insert_command.replace("${i}", str(columnValue.value) + ", ${i}")
|
|
else:
|
|
sql_insert_command = sql_insert_command.replace("${i}", "'" + columnValue.value + "', ${i}")
|
|
|
|
sql_insert_command = sql_insert_command.replace(", ${i}", "")
|
|
return sql_insert_command
|
|
|
|
|
|
def generate_random_column(sql_column: SQLColumn, generated_values: List[ColumnValue], fixed_date_time: bool = False):
|
|
if sql_column.columnType == SQLColumnType.BIGINT:
|
|
return ColumnValue(random.randint(min_bigint, max_bigint), ValueType.INT, sql_column)
|
|
elif sql_column.columnType == SQLColumnType.INT:
|
|
return ColumnValue(random.randint(min_int, max_int), ValueType.INT, sql_column)
|
|
elif sql_column.columnType == SQLColumnType.VARCHAR:
|
|
text_legnth = random.randint(1, 255)
|
|
lorem_text = lorem.sentence()
|
|
return ColumnValue(lorem_text[:text_legnth], ValueType.STRING, sql_column)
|
|
elif sql_column.columnType == SQLColumnType.DATE:
|
|
if "end" in sql_column.columnName or "max" in sql_column.columnName and sql_column.columnType == SQLColumnType.DATE:
|
|
for value in generated_values:
|
|
if ("start" in value.sqlColumn.columnName or "min" in sql_column.columnName) and sql_column.columnType == SQLColumnType.DATE:
|
|
startingDate = datetime.datetime.strptime(value.value, "%Y-%m-%d").date()
|
|
return ColumnValue(str(generate_random_date(startingDate, max_date)), ValueType.STRING, sql_column)
|
|
elif "start" in sql_column.columnName or "min" in sql_column.columnName and sql_column.columnType == SQLColumnType.DATE:
|
|
for value in generated_values:
|
|
if ("end" in value.sqlColumn.columnName or "max" in sql_column.columnName) and sql_column.columnType == SQLColumnType.DATE:
|
|
endingDate = datetime.datetime.strptime(value.value, "%Y-%m-%d").date()
|
|
return ColumnValue(str(generate_random_date(min_date, endingDate)), ValueType.STRING, sql_column)
|
|
return ColumnValue(str(generate_random_date(min_date, max_date)), ValueType.STRING, sql_column)
|
|
elif sql_column.columnType == SQLColumnType.TIME:
|
|
if "end" in sql_column.columnName or "max" in sql_column.columnName and sql_column.columnType == SQLColumnType.TIME:
|
|
for value in generated_values:
|
|
if ("start" in value.sqlColumn.columnName or "min" in sql_column.columnName) and sql_column.columnType == SQLColumnType.TIME:
|
|
startingTime = datetime.datetime.strptime(value.value, "%H:%M:%S").time()
|
|
return ColumnValue(str(generate_random_time(startingTime, max_time)), ValueType.STRING, sql_column)
|
|
elif "start" in sql_column.columnName or "min" in sql_column.columnName and sql_column.columnType == SQLColumnType.TIME:
|
|
for value in generated_values:
|
|
if ("end" in value.sqlColumn.columnName or "max" in sql_column.columnName) and value.sqlColumn.columnType == SQLColumnType.TIME:
|
|
endingTime = datetime.datetime.strptime(value.value, "%H:%M:%S").time()
|
|
return ColumnValue(str(generate_random_time(min_time, endingTime)), ValueType.STRING, sql_column)
|
|
return ColumnValue(str(generate_random_time(min_time, max_time)), ValueType.STRING, sql_column)
|
|
elif sql_column.columnType == SQLColumnType.DATETIME:
|
|
if "end" in sql_column.columnName or "max" in sql_column.columnName and sql_column.columnType == SQLColumnType.DATETIME:
|
|
for value in generated_values:
|
|
if ("start" in value.sqlColumn.columnName or "min" in sql_column.columnName) and value.sqlColumn.columnType == SQLColumnType.DATETIME:
|
|
startingDate = datetime.datetime.strptime(value.value, "%Y-%m-%d %H:%M:%S").date()
|
|
startingTime = datetime.datetime.strptime(value.value, "%Y-%m-%d %H:%M:%S").time()
|
|
if fixed_date_time:
|
|
return ColumnValue(str(generate_random_date_time_fixed_date(startingDate, startingTime, max_time)), ValueType.STRING, sql_column)
|
|
else:
|
|
return ColumnValue(str(generate_random_date_time(startingDate, startingTime, max_date, max_time)), ValueType.STRING, sql_column)
|
|
elif "start" in sql_column.columnName or "min" in sql_column.columnName and sql_column.columnType == SQLColumnType.DATETIME:
|
|
for value in generated_values:
|
|
if ("end" in value.sqlColumn.columnName or "max" in sql_column.columnName) and value.sqlColumn == SQLColumnType.DATETIME:
|
|
endingDate = datetime.datetime.strptime(value.value, "%Y-%m-%d %H:%M:%S").date()
|
|
endingTime = datetime.datetime.strptime(value.value, "%Y-%m-%d %H:%M:%S").time()
|
|
if fixed_date_time:
|
|
return ColumnValue(str(generate_random_date_time_fixed_date(endingDate, min_time, endingTime)), ValueType.STRING, sql_column)
|
|
else:
|
|
return ColumnValue(str(generate_random_date_time(min_date, min_time, endingDate, endingTime)), ValueType.STRING, sql_column)
|
|
return ColumnValue(str(generate_random_date_time(min_date, min_time, max_date, max_time)), ValueType.STRING, sql_column)
|
|
|
|
|
|
def generate_random_date(start_date, end_date):
|
|
delta = end_date - start_date
|
|
random_days = random.randrange(delta.days + 1)
|
|
return start_date + datetime.timedelta(days=random_days)
|
|
|
|
|
|
def generate_random_time(start_time, end_time):
|
|
hours = random.randint(start_time.hour, end_time.hour)
|
|
minutes = random.randint(start_time.minute, end_time.minute)
|
|
seconds = random.randint(start_time.second, end_time.second)
|
|
return datetime.time(hours, minutes, seconds)
|
|
|
|
|
|
def generate_random_date_time(start_date, start_time, end_date, end_time):
|
|
random_date = generate_random_date(start_date, end_date)
|
|
if random_date == start_date:
|
|
random_time = generate_random_time(start_time, end_time)
|
|
else:
|
|
random_time = generate_random_time(min_time, max_time)
|
|
return datetime.datetime.combine(random_date, random_time)
|
|
|
|
|
|
def generate_random_date_time_fixed_date(fixed_date, start_time, end_time):
|
|
random_time = generate_random_time(start_time, end_time)
|
|
return datetime.datetime.combine(fixed_date, random_time)
|