114 lines
3.6 KiB
Python
114 lines
3.6 KiB
Python
import datetime
|
|
import math
|
|
import random
|
|
import string
|
|
import time
|
|
import uuid
|
|
from typing import Any, List, Optional, Sequence, Union
|
|
|
|
|
|
def generate_uuid7() -> str:
|
|
if hasattr(uuid, "uuid7"):
|
|
return str(uuid.uuid7())
|
|
# Fallback implementation compliant with RFC 9562 for UUIDv7
|
|
millis = int(time.time() * 1000)
|
|
rand_a = random.getrandbits(12)
|
|
rand_b = random.getrandbits(62)
|
|
uuid_int = (millis & 0xFFFFFFFFFFFF) << 80
|
|
uuid_int |= (0x7 << 76)
|
|
uuid_int |= (rand_a & 0x0FFF) << 64
|
|
uuid_int |= (0x2 << 62)
|
|
uuid_int |= (rand_b & 0x3FFFFFFFFFFFFFFF)
|
|
return str(uuid.UUID(int=uuid_int))
|
|
|
|
|
|
def datetime_range_generator(days: int, count: int, start_date: str) -> List[str]:
|
|
start = datetime.datetime.strptime(start_date, "%Y-%m-%d")
|
|
base_step = count // days if days > 0 else count
|
|
remainder = count % days if days > 0 else 0
|
|
result: List[str] = []
|
|
for i in range(days):
|
|
day_str = (start + datetime.timedelta(days=i)).strftime("%Y-%m-%d")
|
|
repeats = base_step + (1 if i < remainder else 0)
|
|
result.extend([day_str] * repeats)
|
|
return result
|
|
|
|
|
|
def get_date_before_n_days(n: int) -> str:
|
|
today = datetime.date.today()
|
|
past_date = today - datetime.timedelta(days=n - 1)
|
|
return past_date.strftime("%Y-%m-%d")
|
|
|
|
|
|
def get_random_sublist(
|
|
min_len: int, max_len: int, target: Sequence[Any], rng: random.Random
|
|
) -> List[Any]:
|
|
list_length = len(target)
|
|
if list_length == 0:
|
|
return []
|
|
safe_min = max(0, min(min_len, list_length))
|
|
safe_max = max(safe_min, min(max_len, list_length))
|
|
chosen_size = rng.randint(safe_min, safe_max)
|
|
return rng.sample(list(target), chosen_size)
|
|
|
|
|
|
def sample_weighted_without_replacement(
|
|
items: Sequence[Any], weights: Sequence[float], min_len: int, max_len: int, rng: random.Random
|
|
) -> List[Any]:
|
|
n = len(items)
|
|
if n == 0:
|
|
return []
|
|
safe_min = max(0, min(min_len, n))
|
|
safe_max = max(safe_min, min(max_len, n))
|
|
chosen_size = rng.randint(safe_min, safe_max)
|
|
if chosen_size == 0:
|
|
return []
|
|
if chosen_size >= n:
|
|
return list(items)
|
|
|
|
# Efraimidis-Spirakis algorithm for weighted sampling without replacement
|
|
keys = []
|
|
for i, w in enumerate(weights):
|
|
weight = float(w) if float(w) > 0 else 1e-6
|
|
u = max(1e-10, min(1.0 - 1e-10, rng.random()))
|
|
keys.append((u ** (1.0 / weight), items[i]))
|
|
|
|
keys.sort(key=lambda x: x[0], reverse=True)
|
|
return [item for _, item in keys[:chosen_size]]
|
|
|
|
|
|
def sample_regex_mask(mask: str, rng: random.Random) -> str:
|
|
# '#' -> digit, '?' -> uppercase, '@' -> lowercase, '*' -> alphanumeric
|
|
out = []
|
|
for char in mask:
|
|
if char == "#":
|
|
out.append(rng.choice(string.digits))
|
|
elif char == "?":
|
|
out.append(rng.choice(string.ascii_uppercase))
|
|
elif char == "@":
|
|
out.append(rng.choice(string.ascii_lowercase))
|
|
elif char == "*":
|
|
out.append(rng.choice(string.ascii_letters + string.digits))
|
|
else:
|
|
out.append(char)
|
|
return "".join(out)
|
|
|
|
|
|
def cast_type(value: Any, col_type: str) -> Any:
|
|
if value is None:
|
|
return None
|
|
try:
|
|
if col_type == "integer":
|
|
return int(value)
|
|
elif col_type == "float":
|
|
return float(value)
|
|
elif col_type == "boolean":
|
|
return bool(value)
|
|
elif col_type == "string" or col_type == "uuid7":
|
|
return str(value)
|
|
elif col_type == "datetime":
|
|
return str(value)
|
|
return value
|
|
except (ValueError, TypeError):
|
|
return value
|