from flask import Blueprint, request import requests import logging from datetime import datetime, timedelta from firebase.firebase import Firebase import pytz import utility.function as func import utility.adminPhoneService as adminPhoneService import os import base64 import json from bigquery.bq import BigQuery logging.basicConfig(level=logging.INFO) evntFunction = func.Event() unify = Blueprint('unify', __name__, url_prefix='/unify') fb = Firebase(host=os.environ.get("FIREBASE_HOST")) timezone = pytz.timezone('Asia/Bangkok') @unify.route('/facebook/event/raw', methods=['POST']) @unify.route('/facebook/event/raw/', methods=['POST']) def facebook_main_event(): #date setup dateStr = datetime.now(timezone).date().strftime("%Y%m%d") hourStr = (datetime.now(timezone) - timedelta(hours=1)).strftime("%H") hourStrNow = datetime.now(timezone).strftime("%H") #Explore hook accountMapping = fb.db.reference().child('property').get(shallow=True) listProperty = func.Function.returnPageInListNew(fb, accountMapping, "facebook") logs = fb.db.reference().child(f"log/facebook/hook/{dateStr}/{hourStr}").get() if logs: for log in logs: if logs[log]['object'] == 'page': if 'messaging' in logs[log]['entry'][0]: entry = logs[log]['entry'][0] messageObject = logs[log]['entry'][0]['messaging'][0] pageId = messageObject['recipient']['id'] result = next((item[str(pageId)] for item in listProperty if str(pageId) in item), None) if not result: continue property_id = result['property_id'] psid = messageObject['sender']['id'] env_key, admin_phones, is_exist = adminPhoneService.get_admin_phones(property_id) eventId = func.Function.generate_event_id(psid+str(datetime.now(timezone).timestamp())) eventPack = None #Check phone first if 'message' in messageObject and 'text' in messageObject['message']: userMessage = messageObject['message']['text'] phoneNumber = func.Function.find_phone_number(userMessage) #Add wrong phone number #If spam number pass notSpam = func.Function.clean_spam_phone_number(phoneNumber) if phoneNumber and notSpam: phoneNumberCleaned = func.Function.clean_and_format_thai_phone_number(phoneNumber) if phoneNumberCleaned in admin_phones: logging.info(f"Skip admin phone number: {phoneNumberCleaned} in property: {property_id}") else: phoneNumberHashed = func.Function.hash256(phoneNumberCleaned) phoneObject = [{'key': 'phoneNumber', 'value': phoneNumberHashed}] messageCleaned = func.Function.clean_phone_number_in_message(userMessage, phoneNumber, phoneNumberCleaned) messageCleanedObject = [{'key': 'message', 'value': messageCleaned}] eventPack = evntFunction.socialEventName(eventId=eventId, pageId=pageId, socialId=psid, channel='facebook', eventType='unify', eventName='contact_info_submitted_phone', eventTimeStamp=entry['time'], eventProperty=messageCleanedObject, userProperty=phoneObject, referral=messageObject['message']['referral'] if "referral" in messageObject['message'] else None ) else: messageCleanedObject = [{'key': 'message', 'value': userMessage}] eventPack = evntFunction.socialEventName(eventId=eventId, pageId=pageId, socialId=psid, channel='facebook',eventName='user_message', eventTimeStamp=entry['time'], eventProperty=messageCleanedObject, referral=messageObject['message']['referral'] if 'referral' in messageObject['message'] else None ) elif 'referral' in messageObject: #check ad open eventPack = evntFunction.socialEventName(eventId=eventId, pageId=pageId, socialId=psid, channel='facebook', eventName='click_messenger_ad', eventTimeStamp=entry['time'], referral=messageObject['referral'] ) #Push event if eventPack: fb.db.reference().child(f"event/{dateStr}/{hourStr}/facebook/{property_id}/{str(pageId)}/{eventId}").set(eventPack) return {"status": "success", "message": "Facebook event success"}, 200 return {"status": "success", "message": "Facebook event log no data"}, 200 @unify.route('/allchannel/unify', methods=['POST']) @unify.route('/allchannel/unify/', methods=['POST']) def fcebook_main_unify(): import pandas as pd from bigquery.bq import BigQuery bq = BigQuery() #date setup dateStr = datetime.now(timezone).date().strftime("%Y%m%d") hourStr = (datetime.now(timezone) - timedelta(hours=1)).strftime("%H") mainEvent = fb.db.reference().child(f"event/{dateStr}/{hourStr}").get() for mainChannel in mainEvent: for acc in mainEvent[mainChannel]: for channelId in mainEvent[mainChannel][acc]: for event in mainEvent[mainChannel][acc][channelId]: eventObj = mainEvent[mainChannel][acc][channelId][event] property_id = func.Unify.checkPropertyIdByChannelId(fb, mainChannel, channelId) eventTimeStamp, channel ,eventType, socialId, eventId, pageId, eventName = eventObj['eventTimeStamp'], eventObj['channel'], eventObj['eventType'], eventObj['socialId'], eventObj['eventId'], eventObj['pageId'], eventObj['eventName'] timeStampTransform = datetime.fromtimestamp(eventTimeStamp / 1000.0, timezone).strftime("%Y-%m-%d %H:%M:%S") temp_user_pseudo_id = func.Function.generate_uuid_from_text(socialId) match_user = func.Unify.checkMapping(fb, property_id, temp_user_pseudo_id) refUser = temp_user_pseudo_id isUnify = False if match_user: temp_user_pseudo_id = match_user isUnify = True #check user exist userIdGen = func.Unify.checkUserIdExist(fb, property_id, temp_user_pseudo_id) if not userIdGen: profilePack = func.Profile.pushNewUser(user_pseudo_id=temp_user_pseudo_id, datetime=datetime.now(timezone).isoformat()) fb.db.reference().child(f"account/{acc}/profile/{temp_user_pseudo_id}").set(profilePack) #profile mainChannelPackProfile = func.Profile.pushNewProfile(id=socialId, pageId=pageId, socialId=socialId, channel=mainChannel, datetime=datetime.now(timezone).isoformat()) fb.db.reference().child(f"account/{acc}/profile/{temp_user_pseudo_id}/{mainChannel}").push(mainChannelPackProfile) if eventType == 'standard': #Send data to BQ row = evntFunction.pushEventBigquery( eventId=eventId, eventTimeStamp=timeStampTransform, eventName=eventName, id=socialId, user_pseudo_id=temp_user_pseudo_id, pageId=channelId, source=channel, eventProperty = eventObj['eventProperty'] if 'eventProperty' in eventObj else [], referral= eventObj['referral'] if 'referral' in eventObj else None, ref_user_pseudo_id=refUser ) elif eventType == 'unify': if 'userProperty' in eventObj: if isUnify: for i in eventObj['userProperty']: keyUserProperty = i['key'] valueUserProperty = i['value'] #Check user property checkUserProperty = func.Unify.checkUserProperty(fb, acc, temp_user_pseudo_id, keyUserProperty) if not checkUserProperty: #push user property to profile fb.db.reference().child(f"account/{acc}/profile/{temp_user_pseudo_id}/{keyUserProperty}/").push( func.Profile.pushNewProfile(id=valueUserProperty, channel=mainChannel,pageId=pageId, socialId=socialId, datetime=datetime.now(timezone).isoformat()) ) pass else: #Check exist value or update propertyData = fb.db.reference().child(f"account/{acc}/profile/{temp_user_pseudo_id}/{keyUserProperty}/").get() checkKeyUserUpdate = False for prop in propertyData: valueId = propertyData[prop]['id'] if valueId == valueUserProperty: # findUservalue = valueId findUserPath = prop checkKeyUserUpdate = True else: continue if not checkKeyUserUpdate: fb.db.reference().child(f"account/{acc}/profile/{temp_user_pseudo_id}/{keyUserProperty}/").push( func.Profile.pushNewProfile(id=valueUserProperty, channel=mainChannel, pageId=pageId, socialId=socialId, datetime=datetime.now(timezone).isoformat()) ) fb.db.reference().child(f"account/{acc}/profile/{temp_user_pseudo_id}").update({ 'updated_at': datetime.now(timezone).isoformat() }) else: fb.db.reference().child(f"account/{acc}/profile/{temp_user_pseudo_id}/{keyUserProperty}/{findUserPath}").update({ 'updated_at': datetime.now(timezone).isoformat() }) else: fondUser = False for i in eventObj['userProperty']: keyUserProperty = i['key'] valueUserProperty = i['value'] userReturn = func.Unify.checkUserPropertyTemp(fb, acc, keyUserProperty, valueUserProperty) if userReturn: func.Unify.insertMapping(fb, acc, temp_user_pseudo_id, userReturn) # transfer profile process # Get old profile oldProfileData = fb.db.reference().child(f"account/{acc}/profile/{temp_user_pseudo_id}").get() if oldProfileData: for channelLoop in oldProfileData: if channelLoop not in ['created_at', 'updated_at', 'user_pseudo_id']: for oldLog in oldProfileData[channelLoop]: oldId = oldProfileData[channelLoop][oldLog]['id'] oldPageId = oldProfileData[channelLoop][oldLog]['source']['page_id'] #Loop in old channel newProfileData = fb.db.reference().child(f"account/{acc}/profile/{userReturn}/{channelLoop}").get() isPusholdToNew = True for newLog in newProfileData: newId = newProfileData[newLog]['id'] if str(newId) == str(oldId): isPusholdToNew = False break if isPusholdToNew: mainChannelPackProfile = func.Profile.pushNewProfile(id=oldId, pageId=oldPageId, socialId=oldId, channel=channelLoop, datetime=datetime.now(timezone).isoformat()) fb.db.reference().child(f"account/{acc}/profile/{userReturn}/{channelLoop}").push(mainChannelPackProfile) temp_user_pseudo_id = userReturn fondUser = True break if (fondUser) and (isPusholdToNew != True): #profile mainChannelPackProfile = func.Profile.pushNewProfile(id=socialId, pageId=pageId, socialId=socialId, channel=mainChannel, datetime=datetime.now(timezone).isoformat()) fb.db.reference().child(f"account/{acc}/profile/{temp_user_pseudo_id}/{mainChannel}").push(mainChannelPackProfile) else: for i in eventObj['userProperty']: keyUserProperty = i['key'] valueUserProperty = i['value'] fb.db.reference().child(f"account/{acc}/profile/{temp_user_pseudo_id}/{keyUserProperty}/").push( func.Profile.pushNewProfile(id=valueUserProperty, channel=mainChannel, pageId=pageId, socialId=socialId, datetime=datetime.now(timezone).isoformat()) ) fb.db.reference().child(f"account/{property_id}/profile_temp/{keyUserProperty}/{valueUserProperty}").set(temp_user_pseudo_id) row = evntFunction.pushEventBigquery( eventId=eventId, eventTimeStamp=timeStampTransform, eventName=eventName, id=socialId, user_pseudo_id=temp_user_pseudo_id, pageId=channelId, source=channel, eventProperty = eventObj['eventProperty'] if 'eventProperty' in eventObj else [], referral= eventObj['referral'] if 'referral' in eventObj else None, ref_user_pseudo_id=refUser ) error = bq.load_data(target_table=f"client_{acc}.event", data=[row]) if error: logging.error(f"Error loading data to BigQuery: {error}") else: logging.info(f"Data loaded to BigQuery for eventId: {eventId}") return {"status": "success", "message": "Unify success"}, 200 @unify.route('/line/event/raw', methods=['POST']) @unify.route('/line/event/raw/', methods=['POST']) def line_main_event(): #date setup dateStr = datetime.now(timezone).date().strftime("%Y%m%d") hourStr = (datetime.now(timezone) - timedelta(hours=1)).strftime("%H") accountMapping = fb.db.reference().child('property').get(shallow=True) listLINE = func.Function.returnLINEOAList(fb, accountMapping) logs = fb.db.reference().child(f"log/line/hook/{dateStr}/{hourStr}").get() if logs: for log in logs: channelID = logs[log]['destination'] result = next((item[str(channelID)] for item in listLINE if str(channelID) in item), None) if not result: continue property_id = result['property_id'] env_key, admin_phones, is_exist = adminPhoneService.get_admin_phones(property_id) channelID = result['channel_id'] fb.db.reference().child(f"property/{property_id}/channel/line/{dateStr}/{hourStr}").get() LINE_uid = logs[log]['events'][0]['source']['userId'] eventId = func.Function.generate_event_id(LINE_uid+str(datetime.now(timezone).timestamp())) eventPack = None eventType = logs[log]['events'][0]['type'] timestamp = logs[log]['events'][0]['timestamp'] # event clean process if eventType == "follow": #check follow type isUnblocked = logs[log]['events'][0]['follow']['isUnblocked'] eventName = "LINE_un_block" if isUnblocked else "LINE_add_friend" eventPack = evntFunction.socialEventName(eventId, channelID, LINE_uid, channel="line",eventName=eventName, eventTimeStamp=timestamp) elif eventType=="unfollow": eventPack = evntFunction.socialEventName(eventId, channelID, LINE_uid, channel="line", eventName="LINE_block", eventTimeStamp=timestamp) elif eventType=="beacon": #Check beacon event beaconEventMapping = { 'enter': 'beacon_enter', 'stay': 'beacon_stay', 'banner': 'beacon_banner' } beaconObject = logs[log]['events'][0]['beacon'] beaconType = beaconObject['type'] beaconRefferal = { 'ref': beaconObject['hwid'] } beaconEventName = beaconEventMapping[beaconType] eventPack = evntFunction.socialEventName(eventId, channelID, LINE_uid, channel="line", eventName=beaconEventName, eventTimeStamp=timestamp, referral=beaconRefferal) elif eventType == 'message': # Detect phone messageObject = logs[log]['events'][0]['message'] userMessage = messageObject["text"] phoneNumber = func.Function.find_phone_number(userMessage) #Add wrong phone number #If spam number pass notSpam = func.Function.clean_spam_phone_number(phoneNumber) if phoneNumber and notSpam: phoneNumberCleaned = func.Function.clean_and_format_thai_phone_number(phoneNumber) if phoneNumberCleaned in admin_phones: logging.info(f"Skip admin phone number: {phoneNumberCleaned} in property: {property_id}") else: phoneNumberHashed = func.Function.hash256(phoneNumberCleaned) messageCleaned = func.Function.clean_phone_number_in_message(userMessage, phoneNumber, phoneNumberCleaned) phoneObject = [{'key': 'phoneNumber', 'value': phoneNumberHashed}] messageCleanedObject = [{'key': 'message', 'value': messageCleaned}] eventPack = evntFunction.socialEventName(eventId=eventId, pageId=channelID, socialId=LINE_uid, channel='line', eventType='unify', eventName='contact_info_submitted_phone', eventTimeStamp=timestamp, eventProperty=messageCleanedObject, userProperty=phoneObject ) else: messageCleanedObject = [{'key': 'message', 'value': userMessage}] eventPack = evntFunction.socialEventName(eventId=eventId, pageId=channelID, socialId=LINE_uid, channel='line', eventName='user_message', eventTimeStamp=timestamp, eventProperty=messageCleanedObject, ) #Push event if eventPack: fb.db.reference().child(f"event/{dateStr}/{hourStr}/line/{property_id}/{str(channelID)}/{eventId}").set(eventPack) return {"status": "success", "message": "LINE event success"}, 200 return {"status": "success", "message": "LINE event log no data"}, 200 @unify.route('/real-time/event',methods=['POST']) def real_time_event(): try: envelope = request.get_json() if not envelope: raise ValueError('Invalid Pub/Sub message format') # Pub/Sub messages are base64 encoded message = envelope.get('message') if not message: raise ValueError('No message found in Pub/Sub envelope') data = base64.b64decode(message['data']).decode('utf-8') message_id = message.get('messageId') publish_time = message.get('publishTime') attributes = message.get('attributes') try: parsed_data = json.loads(data) except json.JSONDecodeError: parsed_data = data # If not JSON, keep as string #Start main process ## Event clean process if "social" in parsed_data: datetime_fromat = datetime.now(timezone).isoformat() dateStr = datetime.now(timezone).date().strftime("%Y%m%d") hourStr = (datetime.now(timezone) - timedelta(hours=1)).strftime("%H") accountMapping = fb.db.reference().child('property').get(shallow=True) if parsed_data['social'] == 'line': listLINE = func.Function.returnLINEOAList(fb, accountMapping) channelId = parsed_data['destination'] result = next((item[str(channelId)] for item in listLINE if str(channelId) in item), None) if not result: return {"status": "error", "message": "Channel not found"}, 204 property_id = result['property_id'] channelId = result['channel_id'] event_pack = evntFunction.lineEventManagement(fb, dateStr, hourStr, parsed_data, property_id, channelId) mainChannel = 'line' #Log to Firebase fb.db.reference().child(f"account/{property_id}/log/line/{dateStr}/{hourStr}").push(parsed_data) elif parsed_data['social'] == 'facebook': listProperty = func.Function.returnPageInListNew(fb, accountMapping, "facebook") entry = parsed_data['entry'][0] channelId = None if 'messaging' in entry: messageObject = entry['messaging'][0] channelId = messageObject['recipient']['id'] elif 'changes' in entry: channelId = entry['id'] result = next((item[str(channelId)] for item in listProperty if str(channelId) in item), None) if not result: return {"status": "error", "message": "Channel not found"}, 204 property_id = result['property_id'] event_pack = evntFunction.facebookEventManagement(fb, dateStr, hourStr, parsed_data, property_id, channelId) mainChannel = 'facebook' fb.db.reference().child(f"account/{property_id}/log/facebook/{dateStr}/{hourStr}").push(parsed_data) if event_pack: #Start unify process from bigquery.bq import BigQuery bq = BigQuery() property_id = func.Unify.checkPropertyIdByChannelId(fb, mainChannel, channelId) eventTimeStamp, channel ,eventType, socialId, eventId, pageId, eventName, channelSource = event_pack['eventTimeStamp'], event_pack['channel'], event_pack['eventType'], event_pack['socialId'], event_pack['eventId'], event_pack['pageId'], event_pack['eventName'], event_pack['channel'] # timeStampTransform = datetime.fromtimestamp(eventTimeStamp / 1000.0, timezone).strftime("%Y-%m-%d %H:%M:%S") timeStampTransform = func.Function.transform_timestamp(eventTimeStamp) temp_user_pseudo_id = func.Function.generate_uuid_from_text(socialId) #Reply (this is for mockup test with client) filter_page_id = ["188748337651362"] if pageId in filter_page_id: from utility.demo import facebookMessageMockUp reply_result = facebookMessageMockUp(socialId, pageId) match_user = func.Unify.checkMapping(fb, property_id, temp_user_pseudo_id) refUser = temp_user_pseudo_id isUnify = False if match_user: temp_user_pseudo_id = match_user isUnify = True #check user exist userIdGen = func.Unify.checkUserIdExist(fb, property_id, temp_user_pseudo_id) if not userIdGen: profilePack = func.Profile.pushNewUser(user_pseudo_id=temp_user_pseudo_id, datetime=datetime.now(timezone).isoformat()) fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}").set(profilePack) #profile mainChannelPackProfile = func.Profile.pushNewProfile(id=socialId, pageId=pageId, socialId=socialId, channel=mainChannel, datetime=datetime.now(timezone).isoformat()) fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/{mainChannel}").push(mainChannelPackProfile) #Set Mapping for social ID fb.db.reference().child(f"account/{property_id}/profile_temp/{channelSource}/{socialId}").set(temp_user_pseudo_id) if channelSource == 'line': #Get LINE profile context from utility.line import OA lineOa = OA(access_token=os.environ.get(f"LINE_{pageId}")) profileLINEOA = lineOa.getUserProfile(user_id=socialId) if profileLINEOA: displayName = profileLINEOA.get("displayName", '-') pictureUrl = profileLINEOA.get("pictureUrl", '-') statusMessage = profileLINEOA.get("statusMessage", '-') language = profileLINEOA.get("language", '-') lineProfile = fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/{mainChannel}").get() for line in lineProfile: if lineProfile[line]['id'] == socialId: fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/line/{line}/displayName").set(displayName) fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/line/{line}/pictureUrl").set(pictureUrl) fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/line/{line}/statusMessage").set(statusMessage) fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/line/{line}/language").set(language) if eventType == 'standard': #Send data to BQ row = evntFunction.pushEventBigquery( eventId=eventId, eventTimeStamp=timeStampTransform, eventName=eventName, id=socialId, user_pseudo_id=temp_user_pseudo_id, pageId=channelId, source=channel, eventProperty = event_pack['eventProperty'] if 'eventProperty' in event_pack else [], userProperty= event_pack['userProperty'] if 'userProperty' in event_pack else [], referral= event_pack['referral'] if 'referral' in event_pack else {}, ref_user_pseudo_id=refUser ) fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/updated_at").set(datetime_fromat) elif eventType == 'unify': if 'userProperty' in event_pack: if isUnify: for i in event_pack['userProperty']: keyUserProperty = i['key'] valueUserProperty = i['value'] #Check user property checkUserProperty = func.Unify.checkUserProperty(fb, property_id, temp_user_pseudo_id, keyUserProperty) if not checkUserProperty: #push user property to profile fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/{keyUserProperty}/").push( func.Profile.pushNewProfile(id=valueUserProperty, channel=mainChannel,pageId=pageId, socialId=socialId, datetime=datetime.now(timezone).isoformat()) ) # if f'{keyUserProperty}_PGP' in event_pack: # fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/{keyUserProperty}_PGP/").push( # func.Profile.pushNewProfile(id=event_pack[f'{keyUserProperty}_PGP'], channel=mainChannel,pageId=pageId, socialId=socialId, datetime=datetime.now(timezone).isoformat()) # ) else: #Check exist value or update propertyData = fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/{keyUserProperty}/").get() checkKeyUserUpdate = False for prop in propertyData: valueId = propertyData[prop]['id'] if valueId == valueUserProperty: # findUservalue = valueId findUserPath = prop checkKeyUserUpdate = True else: continue if not checkKeyUserUpdate: fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/{keyUserProperty}/").push( func.Profile.pushNewProfile(id=valueUserProperty, channel=mainChannel, pageId=pageId, socialId=socialId, datetime=datetime.now(timezone).isoformat()) ) # if f'{keyUserProperty}_PGP' in event_pack: # fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/{keyUserProperty}_PGP/").push( # func.Profile.pushNewProfile(id=event_pack[f'{keyUserProperty}_PGP'], channel=mainChannel,pageId=pageId, socialId=socialId, datetime=datetime.now(timezone).isoformat()) # ) fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}").update({ 'updated_at': datetime.now(timezone).isoformat() }) else: fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/{keyUserProperty}/{findUserPath}").update({ 'updated_at': datetime.now(timezone).isoformat() }) #Set user profile temp for next mapping # if '_PGP' not in keyUserProperty: # fb.db.reference().child(f"account/{property_id}/profile_temp/{keyUserProperty}/{valueUserProperty}").set(temp_user_pseudo_id) if 'pgpProperty' in event_pack: for i in event_pack['pgpProperty']: keyPGPProperty = i['key'] valuePGPProperty = i['value'] checkUserProperty = func.Unify.checkUserProperty(fb, property_id, temp_user_pseudo_id, keyPGPProperty) if not checkUserProperty: #push user property to profile fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/{keyPGPProperty}/").push( func.Profile.pushNewProfile(id=valuePGPProperty, channel=mainChannel,pageId=pageId, socialId=socialId, datetime=datetime.now(timezone).isoformat()) ) else: #Check exist value or update propertyData = fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/{keyPGPProperty}/").get() checkKeyUserUpdate = False for prop in propertyData: valueId = propertyData[prop]['id'] if valueId == valuePGPProperty: # findUservalue = valueId findUserPath = prop checkKeyUserUpdate = True else: continue if not checkKeyUserUpdate: fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/{keyPGPProperty}/").push( func.Profile.pushNewProfile(id=valuePGPProperty, channel=mainChannel, pageId=pageId, socialId=socialId, datetime=datetime.now(timezone).isoformat()) ) fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}").update({ 'updated_at': datetime.now(timezone).isoformat() }) else: fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/{keyPGPProperty}/{findUserPath}").update({ 'updated_at': datetime.now(timezone).isoformat() }) else: fondUser = False isPusholdToNew = True for i in event_pack['userProperty']: keyUserProperty = i['key'] valueUserProperty = i['value'] # if '_PGP' in keyUserProperty: # continue userReturn = func.Unify.checkUserPropertyTemp(fb, property_id, keyUserProperty, valueUserProperty) if userReturn: #Check parent user parentUser = func.Unify.checkParentProfile(fb, property_id, userReturn) if parentUser: userReturn = parentUser #Insert mapping func.Unify.insertMapping(fb, property_id, temp_user_pseudo_id, userReturn) logging.info(f"insertMapping: temp: {temp_user_pseudo_id} --> {userReturn}") #Update key to new user fb.db.reference().child(f"account/{property_id}/profile_temp/{keyUserProperty}/{valueUserProperty}").set(userReturn) # transfer profile process # Get old profile oldProfileData = fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}").get() if oldProfileData: for channelLoop in oldProfileData: if channelLoop not in ['created_at', 'updated_at', 'user_pseudo_id']: oldDistID = [] newDistID = [] mappLogOld = {} for oldLog in oldProfileData[channelLoop]: oldId = oldProfileData[channelLoop][oldLog]['id'] oldDistID.append(oldId) mappLogOld.update({oldId: oldLog}) newProfileData = fb.db.reference().child(f"account/{property_id}/profile/{userReturn}/{channelLoop}").get() if newProfileData: for newLog in newProfileData: newId = newProfileData[newLog]['id'] newDistID.append(newId) set_oldDistID = set(oldDistID) set_newDistID = set(newDistID) listNeedtoTransfer = list(set_oldDistID - set_newDistID) #Loop to transfer for pro in listNeedtoTransfer: tempProfile = fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/{channelLoop}/{mappLogOld[pro]}").get() #Push to parent profile fb.db.reference().child(f"account/{property_id}/profile/{userReturn}/{channelLoop}").push(tempProfile) #Assign userReturn to temp_user_pseudo_id temp_user_pseudo_id=userReturn else: #Set profile temp for next time fb.db.reference().child(f"account/{property_id}/profile_temp/{keyUserProperty}/{valueUserProperty}").set(temp_user_pseudo_id) #Add user property to profile fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/{keyUserProperty}/").push( func.Profile.pushNewProfile(id=valueUserProperty, channel=mainChannel, pageId=pageId, socialId=socialId, datetime=datetime.now(timezone).isoformat()) ) # if f'{keyUserProperty}_PGP' in event_pack: # fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/{keyUserProperty}_PGP/").push( # func.Profile.pushNewProfile(id=event_pack[f'{keyUserProperty}_PGP'], channel=mainChannel,pageId=pageId, socialId=socialId, datetime=datetime.now(timezone).isoformat()) # ) if 'pgpProperty' in event_pack: for i in event_pack['pgpProperty']: keyPGPProperty = i['key'] valuePGPProperty = i['value'] checkUserProperty = func.Unify.checkUserProperty(fb, property_id, temp_user_pseudo_id, keyPGPProperty) if not checkUserProperty: #push user property to profile fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/{keyPGPProperty}/").push( func.Profile.pushNewProfile(id=valuePGPProperty, channel=mainChannel,pageId=pageId, socialId=socialId, datetime=datetime.now(timezone).isoformat()) ) else: #Check exist value or update propertyData = fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/{keyPGPProperty}/").get() checkKeyUserUpdate = False for prop in propertyData: valueId = propertyData[prop]['id'] if valueId == valuePGPProperty: # findUservalue = valueId findUserPath = prop checkKeyUserUpdate = True else: continue if not checkKeyUserUpdate: fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/{keyPGPProperty}/").push( func.Profile.pushNewProfile(id=valuePGPProperty, channel=mainChannel, pageId=pageId, socialId=socialId, datetime=datetime.now(timezone).isoformat()) ) fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}").update({ 'updated_at': datetime.now(timezone).isoformat() }) else: fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/{keyPGPProperty}/{findUserPath}").update({ 'updated_at': datetime.now(timezone).isoformat() }) if 'pgpProperty' in event_pack: for pgp in event_pack['pgpProperty']: keyPGPProperty = pgp['key'] valuePGPProperty = pgp['value'] if keyPGPProperty.endswith('_PGP'): event_pack['userProperty'].append({'key': keyPGPProperty, 'value': valuePGPProperty}) row = evntFunction.pushEventBigquery( eventId=eventId, eventTimeStamp=timeStampTransform, eventName=eventName, id=socialId, user_pseudo_id=temp_user_pseudo_id, pageId=channelId, source=channel, eventProperty = event_pack['eventProperty'] if 'eventProperty' in event_pack else [], userProperty= event_pack['userProperty'] if 'userProperty' in event_pack else [], referral= event_pack['referral'] if 'referral' in event_pack else {}, ref_user_pseudo_id=refUser ) #Update Firebase profile updated_at fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/updated_at").set(datetime_fromat) error = bq.load_data(target_table=f"client_{property_id}.event", data=[row]) if error: logging.error(f"Error loading data to BigQuery: {error}") else: logging.info(f"Data loaded to BigQuery for eventId: {eventId}") return {"status": "success", "message": "Real-time event processed successfully"}, 200 return {"status": "error", "message": "No valid social data found in Pub/Sub message"}, 204 return {"status": "error", "message": "No valid social data found in Pub/Sub message"}, 204 except Exception as e: logging.error(f"Error processing Pub/Sub message: {e}") return {"status": "error", "message": str(e)}, 204 @unify.route('/ga4/unify', methods=['POST']) @unify.route('/ga4/unify/', methods=['POST']) def ga4_uniffy_daily_task(): #GA all GA4 Job from bigquery.bq import BigQuery bigqueryClient = BigQuery() propertyWithGA4 = [] allProperty = fb.db.reference().child("property").get(shallow=True) for propertyId in allProperty: allChannel = fb.db.reference().child("property").child(propertyId).child("channel").get(shallow=True) if 'ga4' in allChannel: ga4Context = fb.db.reference().child("property").child(propertyId).child("channel").child("ga4").get() for ga4p in ga4Context: logging.info(ga4p) if "userProperty" in ga4Context[ga4p] and "unified" in ga4Context[ga4p]['userProperty']: contextData = { "property_id" : propertyId, "ga4_context": ga4Context[ga4p] } propertyWithGA4.append(contextData) #Daily unify for ga4p in propertyWithGA4: property_id = ga4p['property_id'] ga4_property_id = ga4p['ga4_context']['id'] unified_field = ga4p['ga4_context']['userProperty']['unified'] unified_field_list = list(unified_field.keys()) mainChannel = "ga4" #Get daily user Data datas = bigqueryClient.get_ga4_user_data(ga4_property_id, '2025-09-12') for data in datas: # logging.info(data) pseudo_user_id = data['pseudo_user_id'] geo = data.get("geo", None) device = data.get("device", None) userPropertyRestruct = {} if len(data['user_properties']) > 0: for u in data['user_properties']: f = {u["value"]['user_property_name']: u["value"]['string_value']} userPropertyRestruct.update(f) logging.info(userPropertyRestruct) logging.info('\n') last_active_timestamp_micros = datetime.fromtimestamp(int(data['user_info']['last_active_timestamp_micros']) / 1000000, timezone).isoformat() temp_user_pseudo_id = func.Function.generate_uuid_from_text(str(pseudo_user_id)) property_id = func.Unify.checkPropertyIdByChannelId(fb, mainChannel, ga4_property_id) match_user = func.Unify.checkMapping(fb, property_id, temp_user_pseudo_id) refUser = temp_user_pseudo_id isUnify = False newnew = False if match_user: temp_user_pseudo_id = match_user isUnify = True # #check user exist # userIdGen = func.Unify.checkUserIdExist(fb, property_id, temp_user_pseudo_id) # if not userIdGen: # profilePack = func.Profile.pushNewUser(user_pseudo_id=temp_user_pseudo_id, datetime=datetime.now(timezone).isoformat()) # fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}").set(profilePack) # #profile # mainChannelPackProfile = func.Profile.pushNewGA4Profile(id=pseudo_user_id, property_id=ga4_property_id, pseudo_user_id=pseudo_user_id, channel=mainChannel, datetime=last_active_timestamp_micros, # user_properties=userPropertyRestruct, geo=geo, device=device) # fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/{mainChannel}").push(mainChannelPackProfile) # newnew = True # else: # logging.info(userIdGen) # allGA4DataOld = fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/ga4").get() if isUnify: #Update non user property # func.Profile.ga4UpdateSameProfile(fb, allGA4DataOld, temp_user_pseudo_id, pseudo_user_id, property_id, geo, device) # Update old profile fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/updated_at").set(datetime.now(timezone).isoformat()) # Update user property if len(data['user_properties']) > 0: for userProp in data['user_properties']: user_property_name = userProp['value']['user_property_name'] user_property_value = userProp['value']['string_value'] for old in allGA4DataOld: oldPack = allGA4DataOld[old] oldGA4ID = oldPack['id'] if oldGA4ID == pseudo_user_id: if user_property_name in oldPack: if user_property_value != oldPack[user_property_name] and user_property_value != '' and user_property_value != None: fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/ga4/{old}/{user_property_name}").set(user_property_value) fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/updated_at").set(datetime.now(timezone).isoformat()) else: #Update non user property func.Profile.ga4UpdateSameProfile(fb, allGA4DataOld, temp_user_pseudo_id, pseudo_user_id, property_id, geo, device) # Update old profile fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/updated_at").set(datetime.now(timezone).isoformat()) #Check user have unified field if len(data['user_properties']) > 0: isCheckAndUpdateSameUser = True allGA4DataOld = fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/ga4").get() for userProp in data['user_properties']: user_property_name = userProp['value']['user_property_name'] user_property_value = userProp['value']['string_value'] if user_property_name not in unified_field_list: continue keyUnify = unified_field[user_property_name] # Find exist key findUser = func.Function.generate_uuid_from_text(str(user_property_value)) checkUser = fb.db.reference().child(f"account/{property_id}/profile/{findUser}").get(shallow=True) if checkUser: #Match case isCheckAndUpdateSameUser = False #Set mapping func.Unify.insertMapping(fb, property_id, temp_user_pseudo_id, checkUser) # Transfer profile ## Get ge4 profile allGA4DataNew = fb.db.reference().child(f"account/{property_id}/profile/{checkUser}/ga4").get() for old in allGA4DataOld: oldPack = allGA4DataOld[old] oldGA4ID = oldPack['id'] for new in allGA4DataNew: newPack = allGA4DataNew[new] newGA4ID = newPack['id'] appendStatus = True if oldGA4ID == newGA4ID: appendStatus = False if appendStatus: fb.db.reference().child(f"account/{property_id}/profile/{checkUser}/ga4").push(oldPack) #Remove website profile ## Wait case : Now dont know how to organize old profile when delete if isCheckAndUpdateSameUser and newnew != True: # Loop over old profile for userProp in data['user_properties']: user_property_name = userProp['value']['user_property_name'] user_property_value = userProp['value']['string_value'] for old in allGA4DataOld: oldPack = allGA4DataOld[old] oldGA4ID = oldPack['id'] if oldGA4ID == pseudo_user_id: if user_property_name in oldPack: if user_property_value != oldPack[user_property_name] and user_property_value != '' and user_property_value != None: fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/ga4/{old}/{user_property_name}").set(user_property_value) fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/updated_at").set(datetime.now(timezone).isoformat()) return {"status": "success", "message": "GA4 unify processed successfully"}, 200 @unify.route('/mapping/user',methods=['POST']) def update_mapping_user(): #Get property try: bq = BigQuery() import pandas as pd properties = fb.db.reference().child("property").get(shallow=True) for prop in properties: allMappingProfile = fb.db.reference().child(f"account/{prop}/mapping").get() if not allMappingProfile: logging.info(f"No mapping profile found for property {prop}") continue output_list = [] for user_pseudo_id, unified_user_pseudo_id in allMappingProfile.items(): output_list.append({ "user_pseudo_id": user_pseudo_id, "unified_user_pseudo_id": unified_user_pseudo_id }) df = pd.DataFrame(output_list) df['lastupdate'] = datetime.now() df['lastupdate'] = pd.to_datetime(df['lastupdate']) #Bigquery task bq.delete_data("customer-360-profile", f"client_{prop}", "profile_mapping_temp") bq.load_data_df(f"client_{prop}", "profile_mapping_temp", df) condition = "ON (ori.user_pseudo_id = temp.user_pseudo_id) " bq.delete_when_match("customer-360-profile", f"client_{prop}", "profile_mapping", f"client_{prop}", "profile_mapping_temp", condition) bq.load_data_df(f"client_{prop}", "profile_mapping", df) return {"status": "success", "message": "Mapping profile processed successfully"}, 200 except Exception as e: return {"status": "fail", "message": str(e)}, 500 @unify.route('/ga4/unify/v2', methods=['POST']) @unify.route('/ga4/unify/v2/', methods=['POST']) def ga4_uniffy_daily_task_v2(): #GA all GA4 Job try: from bigquery.bq import BigQuery bigqueryClient = BigQuery() propertyWithGA4 = [] allProperty = fb.db.reference().child("property").get(shallow=True) for propertyId in allProperty: allChannel = fb.db.reference().child("property").child(propertyId).child("channel").get(shallow=True) if 'ga4' in allChannel: ga4Context = fb.db.reference().child("property").child(propertyId).child("channel").child("ga4").get() for ga4p in ga4Context: logging.info(ga4p) if "userProperty" in ga4Context[ga4p] and "unified" in ga4Context[ga4p]['userProperty']: contextData = { "property_id" : propertyId, "ga4_context": ga4Context[ga4p] } propertyWithGA4.append(contextData) for ga4p in propertyWithGA4: property_id = ga4p['property_id'] #<--C360 property id ga4_property_id = ga4p['ga4_context']['id'] #<--GA4 property id unified_field = ga4p['ga4_context']['userProperty']['unified'] #<--Dict unified_field_list = list(unified_field) unified_field_list_keys = [item['name'] for item in unified_field_list if item['active']] unified_field_dict = { item['name']:item['ref_user_property'] for item in unified_field_list if item['active'] } # unified_field_list = [unified_field_dict.update({item['ref_user_property']: item['name']}) for item in unified_field_list if item['active']] logging.info(unified_field_dict) mainChannel = "ga4" #Get data psesudo from GA4 dateStr = (datetime.now(timezone)-timedelta(days=1)).strftime("%Y-%m-%d") datas = bigqueryClient.get_ga4_user_data(ga4_property_id, dateStr) if not datas: continue for data in datas: # logging.info(data) pseudo_user_id = data['pseudo_user_id'] geo = data.get("geo", None) device = data.get("device", None) user_properties= data.get('user_properties', []) userPropertyRestruct = {} if len(user_properties) > 0: for u in user_properties: user_property_name = u["value"]['user_property_name'] user_property_value = u["value"]['string_value'] # logging.info(f"VAL: {user_property_value}") if user_property_name in unified_field_list_keys: #Get match user findPhone = fb.db.reference().child(f"account/{property_id}/profile_temp/{unified_field_dict[user_property_name]}").order_by_key().start_at(user_property_value).limit_to_first(1).get() findPhone = dict(findPhone) findPhoneKey = list(findPhone.keys()) if str(findPhoneKey[0]).startswith(user_property_value): user_pseudo_id_match = findPhone[findPhoneKey[0]] user_pseudo_id_match_profile = fb.db.reference().child(f"account/{property_id}/profile/{user_pseudo_id_match}").get() if not user_pseudo_id_match_profile: continue if 'ga4' not in user_pseudo_id_match_profile: #Push data to profile mainChannelPackProfile = func.Profile.pushNewGA4Profile( id=pseudo_user_id, channel=mainChannel, property_id=ga4_property_id, pseudo_user_id=pseudo_user_id, datetime=datetime.now(timezone).isoformat(), device=device, geo=geo, user_properties=user_properties ) logging.info("HAVE GA") logging.info(mainChannelPackProfile) fb.db.reference().child(f"account/{property_id}/profile/{user_pseudo_id_match}/ga4").push(mainChannelPackProfile) else: user_pseudo_id_match_profile_ga4 = user_pseudo_id_match_profile['ga4'] for log in user_pseudo_id_match_profile_ga4: context = user_pseudo_id_match_profile_ga4['log'] context_id = context['id'] if context_id == pseudo_user_id: mainChannelPackProfile = func.Profile.pushNewGA4Profile( id=pseudo_user_id, channel=mainChannel, property_id=ga4_property_id, pseudo_user_id=pseudo_user_id, datetime=datetime.now(timezone).isoformat(), device=device, geo=geo, user_properties=user_properties ) logging.info("NON GA4") logging.info(mainChannelPackProfile) fb.db.reference().child(f"account/{property_id}/profile/{user_pseudo_id_match}/ga4/{log}").set(mainChannelPackProfile) else: continue else: continue return {'status': 'success', 'message': 'GA4 Task done'},200 except Exception as e: logging.error(e) return