<gh_stars>1-10
import uuid
import time
import random
import json
import time
from kafka import KafkaProducer
from azure.servicebus import ServiceBusService

### Change to True if using Kafka for ingestion
kafka = False
currentDirectory = os.getcwd()

### EVENT HUB CONFIGURATION
EVENT_HUB_NAMESPACE = ""
EVENT_HUB_NAME = ""
SHARED_ACCESS_KEY_NAME = ""
KEY_VALUE = ""


# KAFKA CONFIGURATION
WORKER_IP_A =  ""
WORKER_IP_B =  ""
TOPIC_NAME = ""

if kafka == False:
    sbs = ServiceBusService(service_namespace=EVENT_HUB_NAMESPACE, shared_access_key_name=SHARED_ACCESS_KEY_NAME, shared_access_key_value=KEY_VALUE)
else:
    producer = KafkaProducer(bootstrap_servers=['',''],value_serializer=lambda v: json.dumps(v).encode('utf-8'))

## Establish list of store IDs
storeids = list(range(1000, 1030))

# Read in list of products sold by the store
with open("{0}\\GROCERY_DATA.csv".format(currentDirectory), "r") as data:
    products = [row[:-1].split(",") for row in data]
    products = products[1:]

## Item ID for Cough Syrup for usage in Cough Syrup Diversion scenario
## Optional -- you need to add logic to randomly insert large purchases of cough syrup to flag
coughSyrupID = 300436344045


# Letters for the construction of the transaction id
letters = 'ABCDEFGHIJKLMNOPQRSTUVWXYZ'
# starting point for the transactions ids generated by this script
transactionIDStart = 81727
# establishing a starting timestamp
timeStamp = int(str(time.time()).split(".")[0])
# generator of transactions
for y in range(0,2):
    # define the number of transactions recorded in this second
    transactionsPerSecond = random.randint(1,20)
    # establish array for holding the products within the transactions
    transactions = []
    # generation of transactions per second
    for transaction in range(0, transactionsPerSecond):
        # Get a letter for the transaction id
        transactionLetter = letters[random.randint(0,25)]
        # get the transaction id number for this transaction
        transactionIDnum = str(transactionIDStart)
        # get a store id for this transaction
        storeID = storeids[random.randint(0, (len(storeids)-1))]
        # number of products bought in this transaction
        productsPerTransaction = random.randint(1,100)
        # array for holding dict objects representing products bought in this transaction
        productsInTransaction = []
        # generator of products per transaction
        for product in range(0, productsPerTransaction):
            product = products[random.randint(0,(len(products)-1))]
            readingJson = {
                'transactionID': str(storeID) + "-" + transactionIDnum + "-" + transactionLetter, 
                'transactiondate': timeStamp,
                'productid': product[2],
                'category': product[3],
                'price': product[-1],
                'brand': product[4],
                'storeId': storeID, 
                'productname': product[6]
            }
            # append the product generated in readingJson to productsInTransaction array
            productsInTransaction.append(readingJson)
        # append an array of product jsons generated aboce to the overall transactions list
        transactions.append(productsInTransaction)
        # increment id start to get new ID number for transaction id construction
        transactionIDStart += 1
    
    if kafka == False:
        # for each transaction generated
        for transaction in transactions:
            # for each product in each transaction
            for reading in transaction:
                # use json.dumps to convert the dictionary to a JSON format
                s = json.dumps(reading)
                # send to Azure Event Hub
                sbs.send_event(EVENT_HUB_NAME, s)
                print(s)
    else:
        # for each transaction generated
        for transaction in transactions:
            # for each product in each transaction
            for reading in transaction:
                s = json.dumps(reading)
                producer.send(TOPIC_NAME, s)
                print(s)
    # delay a second
    time.sleep(1)
    # add secodn to timestamp
    timeStamp += 1

import uuid
import time
import random
import json
import time
from kafka import KafkaProducer
from azure.servicebus import ServiceBusService

### Change to True if using Kafka for ingestion
kafka = False
currentDirectory = os.getcwd()

### EVENT HUB CONFIGURATION
EVENT_HUB_NAMESPACE = ""
EVENT_HUB_NAME = ""
SHARED_ACCESS_KEY_NAME = ""
KEY_VALUE = ""


# KAFKA CONFIGURATION
WORKER_IP_A =  ""
WORKER_IP_B =  ""
TOPIC_NAME = ""

if kafka == False:
    sbs = ServiceBusService(service_namespace=EVENT_HUB_NAMESPACE, shared_access_key_name=SHARED_ACCESS_KEY_NAME, shared_access_key_value=KEY_VALUE)
else:
    producer = KafkaProducer(bootstrap_servers=['',''],value_serializer=lambda v: json.dumps(v).encode('utf-8'))

## Establish list of store IDs
storeids = list(range(1000, 1030))

# Read in list of products sold by the store
with open("{0}\\GROCERY_DATA.csv".format(currentDirectory), "r") as data:
    products = [row[:-1].split(",") for row in data]
    products = products[1:]

## Item ID for Cough Syrup for usage in Cough Syrup Diversion scenario
## Optional -- you need to add logic to randomly insert large purchases of cough syrup to flag
coughSyrupID = 300436344045


# Letters for the construction of the transaction id
letters = 'ABCDEFGHIJKLMNOPQRSTUVWXYZ'
# starting point for the transactions ids generated by this script
transactionIDStart = 81727
# establishing a starting timestamp
timeStamp = int(str(time.time()).split(".")[0])
# generator of transactions
for y in range(0,2):
    # define the number of transactions recorded in this second
    transactionsPerSecond = random.randint(1,20)
    # establish array for holding the products within the transactions
    transactions = []
    # generation of transactions per second
    for transaction in range(0, transactionsPerSecond):
        # Get a letter for the transaction id
        transactionLetter = letters[random.randint(0,25)]
        # get the transaction id number for this transaction
        transactionIDnum = str(transactionIDStart)
        # get a store id for this transaction
        storeID = storeids[random.randint(0, (len(storeids)-1))]
        # number of products bought in this transaction
        productsPerTransaction = random.randint(1,100)
        # array for holding dict objects representing products bought in this transaction
        productsInTransaction = []
        # generator of products per transaction
        for product in range(0, productsPerTransaction):
            product = products[random.randint(0,(len(products)-1))]
            readingJson = {
                'transactionID': str(storeID) + "-" + transactionIDnum + "-" + transactionLetter, 
                'transactiondate': timeStamp,
                'productid': product[2],
                'category': product[3],
                'price': product[-1],
                'brand': product[4],
                'storeId': storeID, 
                'productname': product[6]
            }
            # append the product generated in readingJson to productsInTransaction array
            productsInTransaction.append(readingJson)
        # append an array of product jsons generated aboce to the overall transactions list
        transactions.append(productsInTransaction)
        # increment id start to get new ID number for transaction id construction
        transactionIDStart += 1
    
    if kafka == False:
        # for each transaction generated
        for transaction in transactions:
            # for each product in each transaction
            for reading in transaction:
                # use json.dumps to convert the dictionary to a JSON format
                s = json.dumps(reading)
                # send to Azure Event Hub
                sbs.send_event(EVENT_HUB_NAME, s)
                print(s)
    else:
        # for each transaction generated
        for transaction in transactions:
            # for each product in each transaction
            for reading in transaction:
                s = json.dumps(reading)
                producer.send(TOPIC_NAME, s)
                print(s)
    # delay a second
    time.sleep(1)
    # add secodn to timestamp
    timeStamp += 1
### Change to True if using Kafka for ingestion ### EVENT HUB CONFIGURATION # KAFKA CONFIGURATION ## Establish list of store IDs # Read in list of products sold by the store ## Item ID for Cough Syrup for usage in Cough Syrup Diversion scenario ## Optional -- you need to add logic to randomly insert large purchases of cough syrup to flag # Letters for the construction of the transaction id # starting point for the transactions ids generated by this script # establishing a starting timestamp # generator of transactions # define the number of transactions recorded in this second # establish array for holding the products within the transactions # generation of transactions per second # Get a letter for the transaction id # get the transaction id number for this transaction # get a store id for this transaction # number of products bought in this transaction # array for holding dict objects representing products bought in this transaction # generator of products per transaction # append the product generated in readingJson to productsInTransaction array # append an array of product jsons generated aboce to the overall transactions list # increment id start to get new ID number for transaction id construction # for each transaction generated # for each product in each transaction # use json.dumps to convert the dictionary to a JSON format # send to Azure Event Hub # for each transaction generated # for each product in each transaction # delay a second # add secodn to timestamp