-
Notifications
You must be signed in to change notification settings - Fork 0
/
google_ads.py
95 lines (80 loc) · 3.52 KB
/
google_ads.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
import random
from collections import defaultdict
from threading import Lock
from constants import seed
import logging
import json
random.seed(seed)
class GoogleAds(object):
# Define the types of adverts available
ADVERT_BASIC = 'BASIC'
ADVERT_TARGETED = 'TARGETED'
# Define advert's price
advert_price = {
ADVERT_BASIC: 100,
ADVERT_TARGETED: 200
}
# Google's internal database
users = []
expenses = defaultdict(list)
purchase_history = defaultdict(list)
lock = Lock()
# post an advert about the product
@staticmethod
def post_advertisement(seller, product, advert_type, scale):
# scale of adverts should not be more than number of users
# users = list(set(GoogleAds.users)) # Remove duplicate elements in list
users = list(GoogleAds.users)
scale = min(int(scale), len(users)) # convert scale to integer type
GoogleAds.lock.acquire()
# if advert_type is basic, choose any set of customers
if advert_type == GoogleAds.ADVERT_BASIC:
users = random.choices(GoogleAds.users, k=scale)
users = list(set(users))
users_str = ', '.join(x.name for x in users)
logging.info('[GoogleAds]: Google pushed the %s Ad for product %s of seller %d to users %s ',
advert_type, product.product_name, product.seller_id, users_str)
# if advert_type is targeted, choose user's who were not shown the same advert in previous tick
elif advert_type == GoogleAds.ADVERT_TARGETED:
new_users = list(set(GoogleAds.users) - set(GoogleAds.purchase_history[product]))
if new_users: # if list of new_users is not null
users = random.choices(new_users, k=scale)
users = list(set(users))
users_str = ', '.join(x.name for x in users)
logging.info('[GoogleAds]: Google pushed the %s Ad for product %s of seller %d to user %s ',
advert_type, product.product_name, product.seller_id, users_str)
else:
print("new_users list is null")
else:
print('Not a valid Advert type')
return
# publish the advert to selected user
scale = len(users)
for user in users:
user.view_advert(product)
# update the bill into seller's account
bill = scale * GoogleAds.advert_price[advert_type]
GoogleAds.expenses[seller].append(bill)
# data = json.loads(json.dumps(GoogleAds.expenses))
# logging.info('[GoogleAds]: Google billed the Seller %s ', data)
GoogleAds.lock.release()
# return the bill amount to the seller
return bill
@staticmethod
def register_user(user):
GoogleAds.lock.acquire()
GoogleAds.users.append(user)
# logging.info("[GoogleAds]:Customer %s added to Google list of user with Tolerance:%s",
# user.name, user.tolerance)
GoogleAds.lock.release()
@staticmethod
def track_user_purchase(user, product):
GoogleAds.lock.acquire()
GoogleAds.purchase_history[product].append(user)
GoogleAds.lock.release()
@staticmethod
def user_coverage(product):
# purchase_history = ', '.join(x.name for x in GoogleAds.purchase_history[product])
# print("purchase history"+str(len(GoogleAds.purchase_history[product])))
# print("purchase history"+purchase_history)
return len(set(GoogleAds.purchase_history[product])) / len(GoogleAds.users)