in datasets/thelook_ecommerce/pipelines/_images/run_thelook_kub/fake.py [0:0]
def __post_init__(self, order=None):
global inv_item_id
self.order_id = order.order_id
self.user_id = order.user_id
inv_item_id = inv_item_id + 1
self.inventory_item_id = inv_item_id
self.status = order.status
self.created_at = order.created_at - datetime.timedelta(
seconds=random.randrange(SECONDS_IN_MINUTE * 240)
) # order purchased within 4 hours
self.shipped_at = order.shipped_at
self.delivered_at = order.delivered_at
self.returned_at = order.returned_at
random_idx = np.random.choice(a=len(PRODUCT_GENDER_DICT[order.gender]), size=1)[
0
]
product = PRODUCT_GENDER_DICT[order.gender][random_idx]
self.product_id = product[0]
self.sale_price = product[7]
self.ip_address = fake.ipv4()
self.browser = self.random_item(
population=["IE", "Chrome", "Safari", "Firefox", "Other"],
distribution=[0.05, 0.5, 0.2, 0.2, 0.05],
)
self.traffic_source = self.random_item(
population=["Email", "Adwords", "Organic", "YouTube", "Facebook"],
distribution=[0.45, 0.3, 0.05, 0.1, 0.1],
)
self.session_id = str(uuid.uuid4())
self.person = order.user # pass person object to events
self.is_sold = True
previous_created_at = None
# Generate Events Table
if order.num_of_item == 1: # if only 1 item in order go through flow
for idx, val in enumerate(
["home", "department", "product", "cart", "purchase"]
):
self.sequence_number = idx + 1
self.event_type = val
self.uri = generate_uri(val, product)
events.append(dataclasses.asdict(Events(order_item=self)))
previous_created_at = self.created_at
self.created_at = previous_created_at + datetime.timedelta(
seconds=random.randrange(SECONDS_IN_MINUTE * 3)
)
else: # if multiple items
sequence_num = 0 # track sequence num of purchase event
for _ in range(order.num_of_item):
for event in ["department", "product", "cart"]:
sequence_num += 1
self.sequence_number = sequence_num
self.event_type = event
self.uri = generate_uri(event, product)
events.append(dataclasses.asdict(Events(order_item=self)))
sequence_num = self.sequence_number
previous_created_at = self.created_at
self.created_at = previous_created_at + datetime.timedelta(
seconds=random.randrange(180)
)
self.sequence_number = sequence_num + 1
self.created_at += datetime.timedelta(random.randrange(5))
self.event_type = "purchase"
self.uri = generate_uri("purchase", product)
events.append(dataclasses.asdict(Events(order_item=self)))
# sold inventory item
inventory_items.append(dataclasses.asdict(InventoryItem(order_item=self)))
# unsold inventory items
num_of_items = self.random_item(
population=[1, 2, 3], distribution=[0.5, 0.3, 0.2]
)
for _ in range(num_of_items):
self.is_sold = False
inv_item_id += 1
self.inventory_item_id = inv_item_id
inventory_items.append(dataclasses.asdict(InventoryItem(order_item=self)))