in datasets/thelook_ecommerce/pipelines/_images/run_thelook_kub/fake.py [0:0]
def generate_products() -> typing.List[dict]:
product_brand_dict = {} # products partitioned by brand - unused
product_category_dict = {} # product partitioned by cateogry - unused
gender_category_dict = {} # products partitioned by gender and category - unused
product_id_dict = {} # products to generate events table - unused
product_gender_dict = {} # product partitioned by gender
product_by_id_dict = {} # products partitioned by product ID
products = collections.defaultdict(list)
with open(
f"{os.environ['SOURCE_DIR']}/products.csv", encoding="utf-8"
) as productcsv:
csv_reader = csv.DictReader(productcsv)
for rows in csv_reader:
for k, v in rows.items():
products[k].append(v)
product_id = products["id"]
brands = products["brand"]
name = products["name"]
cost = products["cost"]
category = products["category"]
department = products["department"]
sku = products["sku"]
retail_price = products["retail_price"]
distribution_center_id = products["distribution_center_id"]
for _ in range(len(brands)):
product_brand_dict[brands[_]] = []
product_category_dict[category[_]] = []
product_id_dict[product_id[_]] = []
product_by_id_dict[product_id[_]] = []
if department[_] == "Men":
product_gender_dict["M"] = []
gender_category_dict["M" + category[_]] = []
if department[_] == "Women":
product_gender_dict["F"] = []
gender_category_dict["F" + category[_]] = []
for col in list(
zip(
product_id,
brands,
name,
cost,
category,
department,
sku,
retail_price,
distribution_center_id,
)
):
product_id = col[0]
brand = col[1]
name = col[2]
cost = col[3]
category = col[4]
department = col[5]
sku = col[6]
retail_price = col[7]
distribution_center_id = col[8]
product_by_id_dict[product_id] = {
"brand": brand,
"name": name,
"cost": cost,
"category": category,
"department": department,
"sku": sku,
"retail_price": retail_price,
"distribution_center_id": distribution_center_id,
}
product_brand_dict[brand].append(col)
product_category_dict[category].append(col)
if department == "Men":
product_gender_dict["M"].append(col)
gender_category_dict["M" + category].append(col)
if department == "Women":
product_gender_dict["F"].append(col)
gender_category_dict["F" + category].append(col)
# helper dict to generate events
for col in list(zip(product_id, brands, category, department)):
product_id_dict[col[0]] = {
"brand": col[1],
"category": col[2],
"department": col[3],
}
return product_gender_dict, product_by_id_dict, products