#!pip install Faker

from collections import defaultdict
from faker import Faker
import pandas as pd
import numpy as np
import datetime
import hashlib
import random
import math
import os

from hops import hdfs
from hops import pandas_helper as pandas

# Seed for Reproducibility
faker = Faker()
faker.seed_locale('en_US', 0)

SEED = 123
random.seed(SEED)
np.random.seed(SEED)
faker.seed_instance(SEED)

TOTAL_UNIQUE_TRANSACTIONS = 43200
TOTAL_UNIQUE_USERS = 100
START_DATE = '2021-04-01 00:00:00'
END_DATE = '2021-04-02 00:01:00'
DATE_FORMAT = '%Y-%m-%d %H:%M:%S'

# Change this according to your settings
KAFKA_BROKER_ADDRES = "broker.kafka.service.consul:9091"
KAFKA_TOPIC_NAME = "credit_card_transactions"

def generate_unique_credit_card_numbers(n: int) -> list:
    cc_ids = set()
    for _ in range(n):
        cc_id = faker.credit_card_number(card_type='visa')
        cc_ids.add(cc_id)
    return list(cc_ids) 

credit_card_numbers = generate_unique_credit_card_numbers(TOTAL_UNIQUE_USERS)

assert len(credit_card_numbers) == TOTAL_UNIQUE_USERS 
assert len(credit_card_numbers[0]) == 16 # validate if generated number is 16-digit

# inspect random sample of credit card numbers 
random.sample(credit_card_numbers, 5)

def generate_timestamps(n: int) -> list:
    start = datetime.datetime.strptime(START_DATE, DATE_FORMAT)
    end = datetime.datetime.strptime(END_DATE, DATE_FORMAT)
    timestamps = list()
    for _ in range(n):
        timestamp = faker.date_time_between(start_date=start, end_date=end, tzinfo=None).strftime(DATE_FORMAT)
        timestamps.append(timestamp)
    timestamps = sorted(timestamps)
    return timestamps

timestamps = generate_timestamps(TOTAL_UNIQUE_TRANSACTIONS)

assert len(timestamps) == TOTAL_UNIQUE_TRANSACTIONS

# inspect random sample of timestamps
random.sample(timestamps, 5)

def get_random_transaction_amount(start: float, end: float) -> float:
    amt = round(np.random.uniform(start, end), 2)
    return amt

distribution_percentages = {0.05: (0.01, 1.01), 
                            0.075: (1, 11.01),
                            0.525: (10, 100.01),
                            0.25: (100, 1000.01),
                            0.10: (1000, 10000.01)}

amounts = []

for percentage, span in distribution_percentages.items():
    n = int(TOTAL_UNIQUE_TRANSACTIONS * percentage)
    start, end = span
    for _ in range(n):
        amounts.append(get_random_transaction_amount(start, end+1))
        
random.shuffle(amounts)

assert len(amounts) == TOTAL_UNIQUE_TRANSACTIONS

# inspect random sample of transaction amounts
random.sample(amounts, 5)

def generate_transaction_id(timestamp: str, credit_card_number: str, transaction_amount: float) -> str:
    hashable = f'{timestamp}{credit_card_number}{transaction_amount}'
    hexdigest = hashlib.md5(hashable.encode('utf-8')).hexdigest()
    return hexdigest

transactions = []
for timestamp, amount in zip(timestamps, amounts):
    credit_card_number = random.choice(credit_card_numbers)
    transaction_id = generate_transaction_id(timestamp, credit_card_number, amount)
    transactions.append({'tid': transaction_id, 
                         'datetime': timestamp, 
                         'cc_num': credit_card_number, 
                         'amount': amount, 
                         'fraud_label': 0})

assert len(transactions) == TOTAL_UNIQUE_TRANSACTIONS

# inspect random sample of credit card transactions
random.sample(transactions, 1)

from hops import kafka
from hops import tls
from hops import hdfs
import json
from confluent_kafka import Producer

config = {
    "bootstrap.servers": KAFKA_BROKER_ADDRES,
    "security.protocol": kafka.get_security_protocol(),
    "ssl.ca.location": tls.get_ca_chain_location(),
    "ssl.certificate.location": tls.get_client_certificate_location(),
    "ssl.key.location": tls.get_client_key_location(),
    "group.id": "1"
}

producer = Producer(config)

i = 0
for transaction in transactions:
    if "fraud_label" in transaction:
        transaction.pop("fraud_label")
    if i % 4000 == 0:
        print(json.dumps(transaction))
    producer.produce(KAFKA_TOPIC_NAME, json.dumps(transaction))
    producer.flush()    
    i += 1

FRAUD_RATIO = 0.0025 # percentage of transactions that are fraudulent
NUMBER_OF_FRAUDULENT_TRANSACTIONS = int(FRAUD_RATIO * TOTAL_UNIQUE_TRANSACTIONS)
ATTACK_CHAIN_LENGTHS = [3, 4, 5, 6, 7, 8, 9, 10]

visited = set()
chains = defaultdict(list)

def size(chains: dict) -> int:
    counts = {key: len(values)+1 for (key, values) in chains.items()}
    return sum(counts.values())

def create_attack_chain(i: int):
    chain_length = random.choice(ATTACK_CHAIN_LENGTHS)
    for j in range(1, chain_length):
        if i+j not in visited:
            if size(chains) == NUMBER_OF_FRAUDULENT_TRANSACTIONS:
                break
            chains[i].append(i+j)
            visited.add(i+j)

while size(chains) < NUMBER_OF_FRAUDULENT_TRANSACTIONS:
    i = random.choice(range(TOTAL_UNIQUE_TRANSACTIONS))
    if i not in visited:
        create_attack_chain(i)
        visited.add(i)

assert size(chains) == NUMBER_OF_FRAUDULENT_TRANSACTIONS

def generate_timestamps_for_fraud_attacks(timestamp: str, chain_length: int) -> list:
    timestamps = []
    timestamp = datetime.datetime.strptime(timestamp, DATE_FORMAT)
    for _ in range(chain_length):
        # interval in seconds between fraudulent attacks
        delta = random.randint(30, 120)
        current = timestamp + datetime.timedelta(seconds=delta)
        timestamps.append(current.strftime(DATE_FORMAT))
        timestamp = current
    return timestamps 

def generate_amounts_for_fraud_attacks(chain_length: int) -> list:
    amounts = []
    for percentage, span in distribution_percentages.items():
        n = math.ceil(chain_length * percentage)
        start, end = span
        for _ in range(n):
            amounts.append(get_random_transaction_amount(start, end+1))
    return amounts[:chain_length]

for key, chain in chains.items():
    transaction = transactions[key]
    timestamp = transaction['datetime']
    cc_num = transaction['cc_num']
    amount = transaction['amount']
    transaction['fraud_label'] = 1
    inject_timestamps = generate_timestamps_for_fraud_attacks(timestamp, len(chain))
    inject_amounts = generate_amounts_for_fraud_attacks(len(chain))
    random.shuffle(inject_amounts)
    for i, idx in enumerate(chain):
            original_transaction = transactions[idx]
            inject_timestamp = inject_timestamps[i]
            original_transaction['datetime'] = inject_timestamp
            original_transaction['fraud_label'] = 1
            original_transaction['cc_num'] = cc_num
            original_transaction['amount'] = inject_amounts[i]
            original_transaction['tid'] = generate_transaction_id(inject_timestamp, cc_num, amount)
            transactions[idx] = original_transaction

i = 0
for transaction in transactions:
    if i % 2000 == 0:
        print(json.dumps(transaction))
    producer.produce(KAFKA_TOPIC_NAME, json.dumps(transaction))
    producer.flush()    
    i += 1