import datetime

import json
import os
import psycopg2, [Link]
from itertools import chain
from [Link] import parse_qs
import sys

import pytz

connection = None
cursor_pgsql = None

CUSTOM_OPERATORS = {
"equal_to": "=",
"in": "IN",
"not in": "NOT IN",
"not_equal_to": "!=",
"greater_than": ">",
"less_than": "<",
"greater_than_or_equal_to": ">=",
"less_than_or_equal_to": "<=",
}

# NOTE : In `records_type` field, we can define that module accepts single_record or

multiple_records for particular user.
MODULE_DATA = {
"distributor":{
"target_table": "distributor_master_audience",
"avoid_blocked_records": True,
"blocked_query": """ left join customers on customers.client_entity = 'distributor' and
customers.owner_entity = 'chemist' and customers.owner_entity_id =
current_owner_entity_id and customers.client_entity_id = distributor_master_audience.id
where (customers."status" is null or customers."status" = 'active')"""
}
}

def lambda_handler(event, context):

http_method = event['httpMethod']
path = event['path']

# Routing logic based on the path and method

if "/add_update_preferences" in path and http_method == "POST":
return add_update_preferences_handler(event)
elif "/get_preferences" in path and http_method == "GET":
return get_preferences_handler(event)
 elif "/fetch_preferences_results" in path and http_method == "POST":
return fetch_preferences_results_handler(event)
elif "/fetch_distributor_details" in path and http_method == "POST":
return fetch_distributor_details_handler(event)
elif "/save_preferences" in path and http_method == "POST":
return save_selected_preferences_handler(event)
elif "/get_saved_preferences" in path and http_method == "POST":
return get_saved_preferences_handler(event)
elif "/update_preferences" in path and http_method == "POST":
return update_preferences_handler(event)
elif "/audience_membership" in path and http_method == "POST":
return check_audience_membership_handler(event)
else:
return {
"statusCode": 404,
"body": [Link]({"message": f"API not found-{path}"})
}

def get_db_connection():
try:
global connection
global cursor_pgsql
if connection is None or [Link] != 0:
connection = [Link]([Link]['DATABASE_URL'])
cursor_pgsql = [Link](cursor_factory=[Link])
print("Connection to RDS PostgreSQL database successful.")
return cursor_pgsql, connection
except Exception as e:
return {
"statusCode": 400,
"body": [Link]({"message": f"Error occured in creating DB connection - {e}"})
}

def insert(target_table, columns, values, execute_dict=False, is_multiplicity=False,

db_type="master"):
try:
caller_function = sys._getframe().f_back.f_code.co_name
query = "INSERT INTO " + target_table

columns_template = ",".join(columns)
query += "(" + columns_template + ")"

values_template = ",".join(["%s"] * len(columns))

if is_multiplicity == True:
query += " VALUES (" + values_template + ")"
else:
query += " VALUES (" + values_template + ") RETURNING id"
data = execute_db_query(query, values, is_multiplicity)
 return data
except Exception as e:
return {
"statusCode": 400,
"body": [Link]({"message": f"Error occured in insert record in Database - {e}"})
}

def update(target_table, where_clause, data,

execute_dict=False,params=(),is_multiplicity=False,ignore_str=False):
try:
update_string = ""
for key in [Link]():
if update_string != " ":
update_string += " "
if isinstance(data[key], int) or ignore_str:
data[key] = str(data[key])
update_string += key + " = " + data[key] + ","
else:
data[key] = str(data[key])
data[key] = str(data[key]).replace("%", "%%")
update_string += key + "='" + data[key] + "',"
update_string = update_string.rstrip(update_string[-1])
query = (
"UPDATE "
+ target_table
+ " SET "
+ update_string
+""
+ where_clause
+""
+ "RETURNING id"
)
print("QUERY", query)
if execute_dict:
return execute_db_query_dict(query, params, is_multiplicity)
return execute_db_query(query, params, is_multiplicity)
except Exception as e:
print(e)

def execute_db_query_dict(query, params=None, is_multiquery=False):

connection = None
try:
# Establish a new connection for each request
cursor_pgsql, connection = get_db_connection()

master_cursor = [Link](cursor_factory=[Link])

if is_multiquery:
 master_cursor.executemany(query, params)
else:
master_cursor.execute(query, params)

if (
[Link](" ")[0].lower() == "select"
or [Link](" ")[0].lower() == "with"
):
rec = master_cursor.fetchall()
ans1 = []
for row in rec:
[Link](dict(row))
return ans1
else:
if [Link]().find("returning") > 1:

return master_cursor.fetchone()[0]
[Link]()
return 1

except Exception as e:
print(f"Database error: {e}")
raise

finally:
# Close the cursor and connection after each use
if connection:
cursor_pgsql.close()
[Link]()
print("Database connection closed.")

def execute_db_query(query, params=None, is_multiquery=False):

connection = None
try:
# Establish a new connection for each request
cursor_pgsql, connection = get_db_connection()

master_cursor = [Link](cursor_factory=[Link])

if [Link](" ")[0].upper() == "SELECT":

master_cursor.execute(query, params)
result = master_cursor.fetchall()
else:
if is_multiquery:
master_cursor.executemany(query, params)
else:
master_cursor.execute(query, params)
 if master_cursor.[Link](" ")[0] == "UPDATE":
[Link]()
result = master_cursor.rowcount
elif master_cursor.[Link](" ")[0] == "DELETE":
[Link]()
result = master_cursor.rowcount
elif master_cursor.[Link](" ")[0] == "INSERT":
[Link]()
if [Link]().find("returning") > 1:
result = master_cursor.fetchone()[0]
else:
result = master_cursor.rowcount
elif (
[Link](" ")[0] == "REFRESH"
or [Link](" ")[0] == "TRUNCATE"
):
[Link]()
result = True
else:
result = master_cursor.fetchall()

return result

except Exception as e:
print(f"Database error: {e}")
raise

finally:
# Close the cursor and connection after each use
if connection:
cursor_pgsql.close()
[Link]()
print("Database connection closed.")

def add_update_preferences_handler(event):
try:
body = [Link]([Link]('body', '{}'))
if body["action_type"] == "add":
# Add Query
module = body["module"]
preferences = [Link](body["preferences"])
field_data = [Link](body["field_data"])
add_query = f"INSERT INTO prefrences VALUES ( DEFAULT, '{module}',
{preferences}, {field_data});"
data = execute_db_query_dict(add_query)
result = {
"message": "Preferences added successfully",
"received_data": data
 }
return {
"statusCode": 200,
"body": [Link](result)
}
if body["action_type"] == "update":
# TODO : Have to implement feature to update preferences
pass
except Exception as e:
print(f"Add/Update Preferences Error: {e}")
return {
"statusCode": 400,
"body": [Link]({"message": f"Error occured in Add/Update Preferences - {e}"})
}
def get_preferences_handler(event):
try:
query_params = [Link]('queryStringParameters', {})
print("Query Params", query_params)
if not query_params:
return {
"statusCode": 400,
"body": [Link]({"message": f"Provide valid query parameters"})
}
module = query_params.get('module')
query = f"SELECT id, module, preferences, field_data FROM preferences
WHERE \"module\" = '{module}' AND status = 'active'"
data = execute_db_query_dict(query)
result = {
"message": f"Fetched Preferences for {module}",
"data": data
}

return {
"statusCode": 200,
"body": [Link](result)
}
except Exception as e:
print(f"Get Preferences Error: {e}")
return {
"statusCode": 400,
"body": [Link]({"message": f"Error occured in Get Preferences - {e}"})
}

def fetch_preferences_results_handler(event):
global MODULE_DATA
try:
body = [Link]([Link]('body', '{}'))
if not body or body == {}:
 return {
"statusCode": 400,
"body": [Link]({"message": f"Provide valid body"})
}
module = [Link]('module')
target_table = MODULE_DATA[module]["target_table"]
preferences = [Link]('preferences')
users_list = [Link]('users_list')
query_builder = [Link]('query_builder')
return_query = [Link]('return_query', False)
owner_entity_id = [Link]('owner_entity_id', None)
avoid_blocked_records = MODULE_DATA[module].get("avoid_blocked_records",
False)
if avoid_blocked_records:
blocked_query = MODULE_DATA[module].get("blocked_query")
blocked_query = blocked_query.replace("current_owner_entity_id",
str(owner_entity_id))

else:
blocked_query = None
query = query_builder_lambda(payload=body, avoid_blocked_records =
avoid_blocked_records, blocked_query = blocked_query)
print("------Query---------", query)
if return_query:
return {
"statusCode": 200,
"body": [Link]({"message": f"Fetched Preferences for {target_table}", "query":
query})
}
query_response = execute_db_query_dict(query)
if not query_response:
return {
"statusCode": 400,
"body": [Link]({"message": f"No data found for the given query"})
}

data_list = [item['id'] for item in query_response]

response_data = {
"message": f"Fetched Preferences for {target_table}"
}
response_data["data"] = {
"users_list": data_list,
"total_users": len(data_list)
}
return {
"statusCode": 200,
"body": [Link](response_data)
}
 except Exception as e:
print(f"Fetch Preferences Results Error: {e}")
return {
"statusCode": 400,
"body": [Link]({"message": f"Error occured in Fetch Preferences Results - {e}"})
}

def query_builder_lambda(payload, avoid_blocked_records = False, blocked_query = None):

global CUSTOM_OPERATORS
global MODULE_DATA

preferences = [Link]("preferences", [])

query_builder = [Link]("query_builder", "")
users_list = [Link]("users_list", False)
module = [Link]("module")
target_table = MODULE_DATA.get(module, {}).get("target_table", "default_table") #
Default table if not provided

conditions = {}

for preference in preferences:

ref = preference["ref"].strip()
condition_operator = CUSTOM_OPERATORS.get(preference["condition"])
if not condition_operator:
raise ValueError(f"Invalid condition: {preference['condition']}")

data = preference["data"]
if type(data) is list:
data_list = ", ".join([f"'{item}'" for item in data]) # Format data as SQL-safe strings

# Generate SQL condition

condition = f"{ref} {condition_operator} ({data_list})"
else:
condition = f"{ref} {condition_operator} '{data}'"

# Group by operator if specified

if [Link]("operator") == "OR":
condition = f"({condition})"

# Map the condition to the ID

conditions[str(preference["id"])] = condition

# Replace placeholders in query_builder

final_conditions = query_builder
for key, value in [Link]():
final_conditions = final_conditions.replace(key, value)

# Remove unused placeholders and strip extra spaces

 final_conditions = final_conditions.strip()
if final_conditions.lower() in ("", "none"):
final_conditions = None

# Generate the final query

if users_list:
query = f"SELECT CAST({target_table}.id AS VARCHAR) AS id FROM {target_table}"
else:
query = f"SELECT COUNT({target_table}.id) as count FROM {target_table}"

# Avoid blocked records

if avoid_blocked_records:
query += blocked_query
if final_conditions:
query += f" AND {final_conditions}"
else:
# Add WHERE clause only if final_conditions is valid
if final_conditions:
query += f" WHERE {final_conditions}"

return query

def fetch_distributor_details_handler(event):
global MODULE_DATA
body = [Link]([Link]('body', '{}'))
if not body or body == {}:
return {
"statusCode": 400,
"body": [Link]({"message": f"Provide valid body"})
}
module = [Link]("module")
owner_entity_id = [Link]("owner_entity_id")
distributor_list = [Link]("distributor_list")
target_table = MODULE_DATA[module]['target_table']
if not distributor_list or not owner_entity_id:
return {
"status_code": 400,
"status_message": "Please provide valid required parameters"
}
try:
if not len(distributor_list) > 1:
distributor_list = f'({distributor_list[0]})'
else:
distributor_list = tuple(distributor_list)

existing_distributor_query = f"SELECT [Link], d.distributor_name, [Link], [Link],

d.gstn_number, CASE WHEN d.distributor_software = '' OR d.distributor_software IS NULL
THEN FALSE ELSE TRUE END AS csv_provided, d.distributor_rating FROM ( SELECT
DISTINCT distributor_id AS dist_id FROM chemist_purchases WHERE chemist_id = 681
AND distributor_id IN {distributor_list}) cp LEFT JOIN distributors d ON cp.dist_id = [Link];"

existing_distributor_list = execute_db_query_dict(existing_distributor_query)

if not existing_distributor_list:
existing_distributor_list = []
new_distributor_list = distributor_list
else:
data_list = [str(item['id']) for item in existing_distributor_list]
new_distributor_list = [item for item in list(distributor_list) if item not in data_list]
new_distributor_list = tuple(new_distributor_list)
if not len(new_distributor_list) > 1:
new_distributor_list = f'({new_distributor_list[0]})'
else:
new_distributor_list = tuple(new_distributor_list)
new_distributor_query = f"SELECT id, distributor_name, city, state from {target_table}
where id IN {new_distributor_list}"

new_distributor_list = execute_db_query_dict(new_distributor_query)
print("QUERY", new_distributor_query)
if not new_distributor_list:
return {
"statusCode": 400,
"body": [Link]({"message": f"No data found for the given query"})
}
result = {
"message": "Fetched Distributors list",
"existing_distributors": existing_distributor_list,
"new_distributors": new_distributor_list
}
return {"statusCode":200,
"body":[Link](result)}
except Exception as e:
print(f"Fetch Distributor List Error: {e}")
return {
"statusCode": 400,
"body": [Link]({"message": f"Error occured in Fetch Distributor List - {e}"})
}

def save_selected_preferences_handler(event):
global MODULE_DATA
body = [Link]([Link]('body', '{}'))
target_table = "preference_value"
name = [Link]("name")
owner_entity = [Link]("owner_entity")
owner_entity_id = [Link]("owner_entity_id")
 module = [Link]("module")
selected_preferences = [Link]("selected_preferences")
try:
preference_value_columns = (
"name",
"owner_entity",
"owner_entity_id",
"module",
"selected_preferences"
)
if not (owner_entity or owner_entity_id or module or selected_preferences or name) or
selected_preferences == [] or selected_preferences == {}:
return {
"statusCode": 400,
"body": [Link]({"message": f"Provide all required parameters"})
}
# Perform Insert operations
insert_data = (
name,
owner_entity,
owner_entity_id,
module,
[Link](selected_preferences)
)
preference_insert_response = insert(target_table, preference_value_columns,
insert_data, is_multiplicity=False)
if not preference_insert_response:
return {
"statusCode": 400,
"body": [Link]({"message": f"Error occured in insert preference"})
}
return {
"statusCode": 200,
"body":[Link]({
"message": "Preference Saved/Updated successfully!"
})}
except Exception as e:
print(f"Fetch Distributor List Error: {e}")
return {
"statusCode": 400,
"body": [Link]({"message": f"Error occured in saving preference - {e}"})
}

def get_saved_preferences_handler(event):
global MODULE_DATA
try:
body = [Link]([Link]('body', '{}'))
target_table = "preference_value"
 owner_entity = [Link]("owner_entity")
owner_entity_id = [Link]("owner_entity_id")
module = [Link]("module")
status = [Link]("status")
order_by = [Link]("order_by")
fetch_user_list = [Link]("fetch_user_list", False)

id = [Link]("id")
query = f"""SELECT id, "name", owner_entity,owner_entity_id, "module",
selected_preferences, query_builder, to_char(created_date, 'YYYY-MM-DD"T"HH24:MI:SS')
AS created_date, to_char(updated_date, 'YYYY-MM-DD"T"HH24:MI:SS') AS updated_date,
"status" FROM {target_table}"""
where_clause = f" WHERE owner_entity = '{owner_entity}' AND owner_entity_id =
{owner_entity_id}"
if order_by:
order_by = f" ORDER BY {order_by}"
else:
order_by = " ORDER BY id DESC"
if status:
where_clause += f" AND status = '{status}'"
if id:
where_clause += f" AND id = {id}"
if module:
where_clause += f" AND module = '{module}'"
query += where_clause + order_by
print("QUERY", query)
query_response = execute_db_query_dict(query)
if not query_response:
return {
"statusCode": 200,
"body": [Link]({"message": f"No data found for the given query"})
}

if fetch_user_list:
query_response = query_response[0]
api_payload = {
"module": module,
"users_list": True,
"preferences": query_response['selected_preferences'],
"query_builder": query_response['query_builder']
}
user_list_response = fetch_preferences_results_handler(event =
{"body":[Link](api_payload)})
if user_list_response['statusCode'] == 200:
body = [Link](user_list_response['body'])
result = {
"message": "Fetched user list as per preferences",
"data": body['data']['users_list']
 }
return {"statusCode":200,
"body":[Link](result)}

result = {
"message": "Fetched saved preferences",
"data": query_response
}
return {"statusCode":200,
"body":[Link](result)}
except Exception as e:
return {
"statusCode": 400,
"body": [Link]({"message": f"Error occured in get saved preferences - {e}"})
}

def update_preferences_handler(event):
try:
body = [Link]([Link]('body', '{}'))
if not body or body == {}:
return {
"statusCode": 400,
"body": [Link]({"message": f"Provide valid body"})
}
id = [Link]('id')
if not id or id == {}:
return {
"statusCode": 400,
"body": [Link]({"message": f"Provide valid id"})
}
module = [Link]('module')
selected_preferences = [Link]('selected_preferences')
selected_preferences = [Link](selected_preferences)
name = [Link]('name')
query_builder = [Link]('query_builder')
current_datetime = [Link]([Link]("Asia/Kolkata"))
update_data = {
"selected_preferences": selected_preferences,
"updated_date": current_datetime
}
if name:
update_data['name'] = name
if query_builder:
update_data['query_builder'] = query_builder

where_clause = f"WHERE id = {id} AND module = '{module}'"

 preferences_update_response =
update(target_table="preference_value",where_clause=where_clause,data=update_data)
if not preferences_update_response:
return {
"statusCode": 400,
"body": [Link]({"message": "Error occured in Update Preferences"})
}
return {
"statusCode": 200,
"body": [Link]({"message": "Preferences updated successfully",
"data": preferences_update_response})
}

except Exception as e:
return {
"statusCode": 400,
"body": [Link]({"message": f"Error occured in Update Preferences - {e}"})
}

def get_audience_data(audience_id):
try:
query = f"SELECT * FROM preference_value WHERE id = {audience_id}"
audience_data = execute_db_query_dict(query)
if not audience_data:
return False, "No data found for this "
return True, audience_data
except Exception as e:
return False, "Error occured in get audience data"

def check_audience_membership_handler(event):
try:
body = [Link]([Link]('body', '{}'))
if not body or body == {}:
return {
"statusCode": 400,
"body": [Link]({"message": f"Provide valid body"})
}
module = [Link]('module') # TODO : Might be used in future uses
audience_id = [Link]('audience_id')
if not audience_id:
return {
"statusCode": 400,
"body": [Link]({"message": f"Provide valid audience_id"})
}
membership_data = [Link]('membership_data')
if not membership_data or membership_data == {}:
return {
"statusCode": 400,
 "body": [Link]({"message": f"Provide valid membership_data"})
}

# Fetch Audience data

audience_data_response, audience_data = get_audience_data(audience_id)
if not audience_data_response:
return {
"statusCode": 400,
"body": [Link]({"message": audience_data})
}

preferences = audience_data[0].get('selected_preferences')
query_builder = audience_data[0].get('query_builder')

if 'user' in membership_data:
# Check for Audience from Master Table
body = {
"preferences": preferences,
"module": module,
"users_list": True,
"query_builder": query_builder

}
payload = {"body":[Link](body)}
response = fetch_preferences_results_handler(event = payload)
if response['statusCode'] == 400:
return{
"statusCode": 400,
"body": f"Error occured while fetching results for {module}"
}
if response['statusCode'] == 200:
response_body = [Link](response['body'])
data = response_body['data']
audience_list = data['users_list']
if membership_data['user'] in audience_list:
return {
"statusCode": 200,
"body": [Link]({"message": "Audience Membership Verified",
"status":True, "data": f"{module} - {membership_data['user']} exists in audience"})
}
else:
return {
"statusCode": 200,
"body": [Link]({"message": "Audience Membership Verified",
"status":False, "data": f"{module} - {membership_data['user']} does not exist in audience"})
}
else:
data_found = False
 for data in audience_data:
key = next(iter(membership_data))
value = membership_data[key]
preferences_data = data['selected_preferences']
for preference in preferences_data:
if preference['name'] == key and value in preference['data']:
data_found = True
return {
"statusCode": 200,
"body": [Link]({"message": "Audience Membership Verified",
"status":True, "data": f"{key} - {value} exists in audience"})
}
if not data_found:
return {
"statusCode": 200,
"body": [Link]({"message": "Audience Membership Verified",
"status":False, "data": f"{key} - {value} does not exists in audience"})
}

except Exception as e:
return {
"statusCode": 400,
"body": [Link]({"message": f"Error occured in Check Audience Membership -
{e}"})
}