from flask import Blueprint, request import requests import logging from datetime import datetime from firebase.firebase import Firebase import pytz import utility.function as func import os from bigquery.bq import BigQuery from feature.audience.craft import BQCraft, Query from feature.audience.queryBuilder import * from feature.audience.node import * logging.basicConfig(level=logging.INFO) evntFunction = func.Event() feature_audience = Blueprint('feature_audience', __name__, url_prefix='/feature/audience') fb = Firebase(host=os.environ.get("FIREBASE_HOST")) timezone = pytz.timezone('Asia/Bangkok') @feature_audience.route('/preset/estimate', methods=['GET']) def feature_audience_preset_estimate(): try: if request.method == "GET": data = request.args for req in ['property_id', 'query']: if req not in data: return {'message': f"{req} is required"}, 400 query = data.get("query", None) logging.info(query) #Biquery get data bq = BigQuery() try: df = bq.get_query_df(query) except Exception as e: return {"status": "error", "message": e}, 500 columns = df.columns.to_list() if 'user_pseudo_id' not in columns: return {'status': 'error', 'message': 'user_pseudo_id must be selected in query (use DISTINCT function)'} user_pseudo_id_list = list(df['user_pseudo_id'].unique()) audience_size = len(df['user_pseudo_id'].unique()) return {'status': 'ok', 'audience_size': audience_size, "user_pseudo_id_list": user_pseudo_id_list}, 200 except Exception as e: logging.error(e) return {"status": "error", "message": e}, 500 @feature_audience.route('/preset', methods=['POST','GET', 'PUT']) def feature_audience_preset(): try: if request.method == "POST": data = request.get_json() for req in ['property_id', 'user_id', 'audience_name', 'query', 'description', 'channel']: if req not in data: return {'message': f"{req} is require"}, 400 dateNowStr = datetime.now().strftime("%Y-%m-%d %H:%M:%S") timestampNow = datetime.now().timestamp() property_id = data.get("property_id", None) user_id = data.get("user_id", "No user") name = data.get("audience_name", None) description = data.get("description", None) query = data.get("query", None) channel = data.get("channel") if channel not in ['all', 'line', 'facebook', 'sms', 'email']: return {'status': 'error', 'message': f"not support {channel} for channel parameter"}, 400 #Gen audience id nowStr = str(datetime.now().timestamp()) audienceID = func.Function.generate_uuid_from_text(f"{nowStr}_{name}") #Biquery get data bq = BigQuery() try: df = bq.get_query_df(query) except Exception as e: return {"status": "error", "message": e}, 500 user_pseudo_id_list = list(df['user_pseudo_id'].unique()) audience_size = len(df['user_pseudo_id'].unique()) #Insert Audience context audiences_audience_log_json = BQCraft.audiences_audience_log_json(user_id, "CREATE", query, int(timestampNow)) audiences_json = BQCraft.audiences_json(dateNowStr, dateNowStr, audienceID, name, description, query, user_pseudo_id_list, audience_size, [audiences_audience_log_json]) #Load to Firebase fb.db.reference().child(f"account/{property_id}/audience/{audienceID}").set(audiences_json) fb.db.reference().child(f"account/{property_id}/audience/{audienceID}/status").set('inactive') fb.db.reference().child(f"account/{property_id}/audience/{audience_id}/type").set("preset") # #Insert audience context # error = bq.load_data(f"client_{property_id}.audience", [audiences_json]) # if error: # logging.error(f"Error loading data to BigQuery: {error}") # else: # logging.info(f"Data loaded to BigQuery for Audience ID: {audienceID}") return {'status': 'ok', "message": f"Audience {name} was created successfully", "audience_id": audienceID, "audience_size": audience_size}, 200 elif request.method == "GET": data = request.args for req in ['property_id', 'audience_id']: if req not in data: return {'message': f"{req} is required"}, 400 property_id = data.get("property_id", None) audience_id = data.get("audience_id", None) #Can be all if they want a list of audience logging.info(f"{property_id}_{audience_id}") if audience_id != 'all': audience_context = fb.db.reference().child(f"account/{property_id}/audience/{audience_id}").get() returnAudience = [audience_context] else: audience_context = fb.db.reference().child(f"account/{property_id}/audience").get() if audience_context: returnAudience = [] for a in audience_context: au = audience_context[a] returnAudience.append(au) else: return {'status': 'not fond', 'message': f'Propery {property_id} has no audience'}, 400 # #Get data from bq # bq = BigQuery() # query = Query.get_audience_by_id(property_id, audience_id) # try: # df = bq.get_query_df(query) # except Exception as e: # return {"status": "error", "message": e}, 500 # df_json = df.to_dict(orient='records') # logging.info(df_json) return {'status': 'ok', 'data': returnAudience}, 200 elif request.method == "PUT": data = request.get_json() for req in ['property_id', 'user_id', 'audience_id', 'query']: if req not in data: return {'message': f"{req} is require"}, 400 dateNowStr = datetime.now().strftime("%Y-%m-%d %H:%M:%S") timestampNow = datetime.now().timestamp() # Allow value ['audience_name', 'description', 'query'] property_id = data.get("property_id", None) user_id = data.get("user_id", "No user") audience_id = data.get("audience_id", None) query = data.get("query", None) logging.info(f"{property_id}_{user_id}_{audience_id}") logging.info(query) # #Get context from Bigquery # bq = BigQuery() # query_audience_by_id = Query.get_audience_by_id(property_id, audience_id) # data = bq.get_query(query_audience_by_id) #get context from Firebase createdate = fb.db.reference().child(f"account/{property_id}/audience/{audience_id}/createdate").get() audience_name = fb.db.reference().child(f"account/{property_id}/audience/{audience_id}/audience_name").get() if "audience_name" not in data else data.get("audience_name") description = fb.db.reference().child(f"account/{property_id}/audience/{audience_id}/description").get() if "description" not in data else data.get("description") audience_log = fb.db.reference().child(f"account/{property_id}/audience/{audience_id}/audience_log").get() logging.info("get context from Firebase") #New query bq = BigQuery() try: df = bq.get_query_df(query) except Exception as e: return {"status": "error", "message": e}, 500 logging.info(f"{df.shape[0]}") user_pseudo_id_list = list(df['user_pseudo_id'].unique()) audience_size = len(df['user_pseudo_id'].unique()) #Craft new audiences_audience_log_json = BQCraft.audiences_audience_log_json(user_id, "CHANGE", query, int(timestampNow)) #Append new log audience_log.append(audiences_audience_log_json) logging.info("Append") audiences_json = BQCraft.audiences_json(createdate, dateNowStr, audience_id, audience_name, description, query, user_pseudo_id_list, audience_size, audience_log) logging.info(audiences_json) # df_new = pd.DataFrame([audiences_json]) # df_new['lastupdate'] = pd.to_datetime(df_new['lastupdate']) # df_new['createdate'] = pd.to_datetime(df_new['createdate']) # bq.delete_data("customer-360-profile", f"client_{property_id}", "audience_temp") # bq.load_data_df(f"client_{property_id}", "audience_temp", df_new) # condition = " ON (ori.audience_id = temp.audience_id) " # bq.delete_when_match("customer-360-profile", f"client_{property_id}", "audience", f"client_{property_id}", "audience_temp", condition) # bq.load_data_df(f"client_{property_id}", "audience", df_new) #Update firebase fb.db.reference().child(f"account/{property_id}/audience/{audience_id}/lastupdate").set(dateNowStr) fb.db.reference().child(f"account/{property_id}/audience/{audience_id}/audience_syntax").set(query) fb.db.reference().child(f"account/{property_id}/audience/{audience_id}/audience_name").set(audience_name) fb.db.reference().child(f"account/{property_id}/audience/{audience_id}/description").set(description) fb.db.reference().child(f"account/{property_id}/audience/{audience_id}/audience_log").set(audience_log) fb.db.reference().child(f"account/{property_id}/audience/{audience_id}/audience_size").set(audience_size) fb.db.reference().child(f"account/{property_id}/audience/{audience_id}/user_pseudo_id").set(user_pseudo_id_list) #Process cache profile userContext = [] for user in user_pseudo_id_list: # Get profile userContext.append(fb.db.reference().child(f"account/{property_id}/profile/{user}").get()) logging.info("Process cache profile") finalUser = func.Function.prepcacheAudienceProfile(userContext) logging.info("Process cache profile: 1") df = func.Function.prepcacheAudienceProfileFlattened(finalUser) logging.info("Process cache profile: 2") sorted_json = df.to_dict(orient='records') logging.info("Process cache profile: 3") fb.db.reference().child(f"account/{property_id}/audience/{audience_id}/cache/profile").set(sorted_json) logging.info("Process cache profile: 4") logging.info("Update") return {'status': 'ok', "message": f"Audience {audience_name} was update successfully", "audience_id": audience_id, "audience_size": audience_size}, 200 except Exception as e: logging.error(e) return {"status": "error", "message": e}, 500 return {'status': 'error', 'message': 'Method not allowed'}, 405 @feature_audience.route('/preset/status', methods=['POST']) def feature_audience_preset_status(): try: if request.method == 'POST': data = request.get_json() for req in ['property_id', 'user_id', 'audience_id', 'status']: if req not in data: return {'message': f"{req} is require"}, 400 property_id = data.get("property_id", None) user_id = data.get("user_id", "No user") audience_id = data.get("audience_id", None) status = data.get("status", None) if status not in ['active', 'inactive']: return {'message': f'''status value must equal "active" or "inactive"'''}, 400 #Check audience exist audienceExist = fb.db.reference().child(f"account/{property_id}/audience/{audience_id})").get(shallow=True) if not audienceExist: return {'status': 'not fond', 'message': f'Audience Id {audience_id} not fond'},400 fb.db.reference().child(f"account/{property_id}/audience/{audience_id}/status").set(status) return {'status': 'ok', 'message': f'Change status audience {audience_id} has been change status to {status}'},200 return {'status': 'error', 'message': 'Method not allowed'}, 405 except Exception as e: logging.error(e) return {"status": "error", "message": e}, 500 @feature_audience.route('/preset/recalculate', methods=['POST']) def feature_audience_preset_recalculate(): try: #Get active data from bigquery.bq import BigQuery import numpy as np bq = BigQuery() if request.method == 'POST': accounts = fb.db.reference().child(f"account").get(shallow=True) for acc in accounts: tempAudience = fb.db.reference().child(f"account/{acc}/audience").get(shallow=True) if not tempAudience: continue for au in tempAudience: if not au: continue status = fb.db.reference().child(f"account/{acc}/audience/{au}/status").get() auType = fb.db.reference().child(f"account/{acc}/audience/{au}/type").get() if status == 'active': if auType == 'preset': query = fb.db.reference().child(f"account/{acc}/audience/{au}/audience_syntax").get() #Execute query BQ try: df = bq.get_query_df(query) except Exception as e: return {"status": "error", "message": e}, 500 logging.info(f"{df.shape[0]}") user_pseudo_id_list = list(df['user_pseudo_id'].unique()) audience_size = len(df['user_pseudo_id'].unique()) # Update data fb.db.reference().child(f"account/{acc}/audience/{au}/audience_size").set(audience_size) fb.db.reference().child(f"account/{acc}/audience/{au}/user_pseudo_id").set(user_pseudo_id_list) # fb.db.reference().child(f"account/{acc}/audience/{au}/lastupdate").set(user_pseudo_id_list) #Process cache profile userContext = [] for user in user_pseudo_id_list: # Get profile userContext.append(fb.db.reference().child(f"account/{acc}/profile/{user}").get()) finalUser = func.Function.prepcacheAudienceProfile(userContext) df = func.Function.prepcacheAudienceProfileFlattened(finalUser) sorted_json = df.to_dict(orient='records') fb.db.reference().child(f"account/{acc}/audience/{au}/cache/profile").set(sorted_json) elif auType == 'builder': audienceJson = fb.db.reference().child(f"account/{acc}/audience/{au}/json").get() for node in audienceJson['nodes']: if node['type'] == 'query': query = QueryGenerator(acc, node['query']).generate() node['query'] = query bq = BigQuery() au = AudienceBuilder(acc, au, audienceJson, bq) result = au.execute() if result['status'] == 'ok': resultData = result['data'] audience_size = resultData['final_size'] user_pseudo_id_list = resultData['user_pseudo_id'] fb.db.reference().child(f"account/{acc}/audience/{au}/audience_size").set(audience_size) fb.db.reference().child(f"account/{acc}/audience/{au}/user_pseudo_id").set(user_pseudo_id_list) else: logging.error(result) return {'status': 'ok', 'message': f'All audience has been recalculated'}, 200 return {'status': 'error', 'message': 'Method not allowed'}, 405 except Exception as e: logging.error(e) return {"status": "error", "message": e}, 500 @feature_audience.route('/preset/download', methods=['GET']) def feature_audience_preset_download(): try: if request.method == 'GET': data = request.args for req in ['property_id', 'audience_id', 'row_start', 'row_end']: if req not in data: return {'message': f"{req} is required"}, 400 property_id = data.get("property_id", None) audience_id = data.get("audience_id", None) row_start = data.get("row_start", None) row_end = data.get("row_end", None) #Set row_start = int(row_start) - 1 if int(row_start) > 0 else 0 row_end = int(row_end) data = fb.db.reference().child(f"account/{property_id}/audience/{audience_id}/cache/profile").get() finalData = data[row_start:row_end] return {'status': 'ok', 'data': finalData, 'total_audience': len(data)}, 200 return {'status': 'error', 'message': 'Method not allowed'}, 405 except Exception as e: logging.error(e) return {"status": "error", "message": e}, 500