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 from bigquery.bq import BigQuery logging.basicConfig(level=logging.INFO) evntFunction = func.Event() bq = BigQuery() event = Blueprint('event', __name__, url_prefix='/event') fb = Firebase(host=os.environ.get("FIREBASE_HOST")) timezone = pytz.timezone('Asia/Bangkok') @event.route('/facebook', methods=['POST']) @event.route('/facebook/', methods=['POST']) def main_event(): 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") eventTemp = fb.db.reference().child(f"event/{dateStr}/{hourStr}").get() for ch in eventTemp: for acc in eventTemp[ch]: pageProfile = fb.db.reference().child(f"account/{acc}/profile").get() for page in eventTemp[ch][acc]: pagetemp = eventTemp[ch][acc][page] # pageProfile = fb.db.reference().child(f"account/{acc}/{page}/profile").get() for eventId in pagetemp: eventData = pagetemp[eventId] socialId = eventData['socialId'] channel = eventData['channel'] eventName = eventData['eventName'] eventTimeStamp = eventData['eventTimeStamp'] message = eventData['message'] if 'message' in eventData else None phoneNumber = eventData['phoneNumber'] if 'phoneNumber' in eventData else None referral = eventData['referral'] if 'referral' in eventData else None temp_user_pseudo_id = func.Function.generate_uuid_from_text(socialId) timeStampTransform = datetime.fromtimestamp( int(datetime.now().timestamp()), timezone ).strftime("%Y-%m-%d %H:%M:%S") #Get profile data profileData = fb.db.reference().child(f"account/{acc}/profile/{temp_user_pseudo_id}").get() if not profileData: check = False for pro in pageProfile: facebookPro = pageProfile[pro]['facebook'] for fbProfile in facebookPro: if int(fbProfile['id']) == int(socialId): temp_user_pseudo_id = socialId check = True break if check: #BQ process row = func.Event.pushEventBigquery( eventId=eventId, eventTimeStamp=timeStampTransform, eventName=eventName, id=socialId, user_pseudo_id=temp_user_pseudo_id, pageId=page, source=channel, eventProperty = ([{'key': 'message', 'value': message}] * bool(message) + [{'key': 'phoneNumber', 'value': phoneNumber}] * bool(phoneNumber)) or [], referral=referral if referral else None ) error = bq.load_data(target_table=f"client_{acc}.event", data=[row]) if error: logging.error(f"Error loading data to BigQuery: {error} | {eventId}") else: logging.info(f"Data loaded to BigQuery for eventId: {eventId}") else: #Generate new user and profile #Check phone id # Improssible case # Noti to Google chat noti = func.Notify(key=os.environ.get("GOOGLE_CHAT_NOTI_KEY")) noti.googleChatNotify(message=f"Error Logic: No profile data found for eventId: {eventId} with socialId: {socialId} in account: {acc} and page: {page}. Event Name: {eventName}. Please check the logic.") else: #BQ process #craft data row = func.Event.pushEventBigquery( eventId=eventId, eventTimeStamp=timeStampTransform, eventName=eventName, id=socialId, user_pseudo_id=temp_user_pseudo_id, pageId=page, source=channel, eventProperty = [{'key': 'message', 'value': message}] * bool(message) + [{'key': 'phoneNumber', 'value': phoneNumber}] * bool(phoneNumber) or [], referral=referral if referral else None ) 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": "Facebook event success"}, 200 @event.route('/webhook/', methods=['POST']) @event.route('/webhook//', methods=['POST']) def webhook_event_unify(hook_id): if request.method == "POST": data = request.get_json() for req in ['propertyId', 'eventName', 'socialId', 'channel']: if req not in data: return {'message': f"{req} is required"}, 400 property_id = data.get('propertyId') env_key, admin_phones, is_exist = adminPhoneService.get_admin_phones(property_id) social_id = data.get('socialId') eventName = data.get("eventName") # events = data.get('events') source = data.get('channel') pageId = data.get('pageId', '-') eventProperty = data.get('eventProperty', []) userProperty = data.get('userProperty', []) #PGP process userPropertyCleaned = [] pgpPropertyCleaned = [] if userProperty: pubkey = func.Key.load_key_from_env("public_key") for up, item in enumerate(userProperty[:]): keyUserProperty = item['key'] valueUserProperty = item['value'] if keyUserProperty == 'phoneNumber': phoneNumber = func.Function.find_phone_number(valueUserProperty) # notSpam = func.Function.clean_spam_phone_number(phoneNumber) if phoneNumber: for ph in phoneNumber: notSpam = func.Function.clean_spam_phone_number(ph) if not notSpam: continue phoneNumberCleaned = func.Function.clean_and_format_thai_phone_number(ph) if phoneNumberCleaned in admin_phones: logging.info(f"Skip admin phone number: {phoneNumberCleaned} in property: {property_id}") else: phoneNumberHashed = func.Function.hash256(phoneNumberCleaned) phonePGP = func.Key.pgp_encrypt(phoneNumberCleaned, pubkey) phonePGPPack = {'key': 'phoneNumber_PGP', 'value': phonePGP} phoneHashPack = {'key': 'phoneNumber', 'value': phoneNumberHashed} pgpPropertyCleaned.append(phonePGPPack) userPropertyCleaned.append(phoneHashPack) #update value to hashed # data['userProperty'][up] = {'key': 'phoneNumber', 'value': phoneNumberHashed} # data.update({'phoneNumber_PGP': phonePGP}) else: userProperty.pop(up) elif keyUserProperty == 'email': emailHashed = func.Function.hash256(valueUserProperty) emailPGP = func.Key.pgp_encrypt(valueUserProperty, pubkey) emailPGPPack = {'key': 'email_PGP', 'value': emailPGP} emailHashPack = {'key': 'email', 'value': emailHashed} pgpPropertyCleaned.append(emailPGPPack) userPropertyCleaned.append(emailHashPack) elif keyPGPProperty in ['line', 'facebook']: userPropertyCleaned.append({'key': keyPGPProperty, 'value': valueUserProperty}) # data['userProperty'][up] = {'key': 'email', 'value': emailHashed} # data.update({'email_PGP': emailPGP}) userProperty = userPropertyCleaned data['userProperty'] = userPropertyCleaned data['pgpProperty'] = pgpPropertyCleaned dateStr = datetime.now(timezone).date().strftime("%Y%m%d") hourStr = (datetime.now(timezone)).strftime("%H") timeStampTransform = datetime.fromtimestamp( int(datetime.now().timestamp()), timezone ).strftime("%Y-%m-%d %H:%M:%S") eventId = func.Function.generate_event_id(social_id+str(datetime.now(timezone).timestamp())+hook_id) temp_user_pseudo_id = func.Function.generate_uuid_from_text(str(social_id).strip()) match_user = func.Unify.checkMapping(fb, property_id, temp_user_pseudo_id) refUser = temp_user_pseudo_id isUnified = False if match_user: temp_user_pseudo_id = match_user isUnified = True 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=social_id, pageId=pageId, socialId=social_id, channel=source, datetime=datetime.now(timezone).isoformat()) fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/{source}").push(mainChannelPackProfile) #Set Mapping for social ID fb.db.reference().child(f"account/{property_id}/profile_temp/{source}/{social_id}").set(temp_user_pseudo_id) if source == '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=social_id) 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}/{source}").get() for line in lineProfile: if lineProfile[line]['id'] == social_id: 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 not isUnified: if userProperty: for i in userProperty: keyUserProperty = i['key'] valueUserProperty = i['value'] if keyUserProperty in ['phoneNumber', 'email', 'facebook', 'line', 'phoneNumber_PGP', 'email_PGP']: # findMatching userReturn = func.Unify.checkUserPropertyTemp(fb, property_id, keyUserProperty, valueUserProperty) if userReturn: #Match case *** #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 #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=source, pageId=pageId, socialId=social_id, datetime=datetime.now(timezone).isoformat()) ) # if f'{keyUserProperty}_PGP' in data: # fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/{keyUserProperty}_PGP/").push( # func.Profile.pushNewProfile(id=data[f'{keyUserProperty}_PGP'], channel=source,pageId=pageId, socialId=social_id, datetime=datetime.now(timezone).isoformat()) # ) if 'pgpProperty' in data: for i in data['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=source,pageId=pageId, socialId=social_id, 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=source, pageId=pageId, socialId=social_id, 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: if userProperty: for i in userProperty: keyUserProperty = i['key'] valueUserProperty = i['value'] if keyUserProperty in ['phoneNumber', 'email', 'facebook', 'line', 'phoneNumber_PGP', 'email_PGP']: #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=source,pageId=pageId, socialId=social_id, datetime=datetime.now(timezone).isoformat()) ) # if f'{keyUserProperty}_PGP' in data: # fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/{keyUserProperty}_PGP/").push( # func.Profile.pushNewProfile(id=data[f'{keyUserProperty}_PGP'], channel=source,pageId=pageId, socialId=social_id, 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=source, pageId=pageId, socialId=social_id, datetime=datetime.now(timezone).isoformat()) ) # if f'{keyUserProperty}_PGP' in data: # fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/{keyUserProperty}_PGP/").push( # func.Profile.pushNewProfile(id=data[f'{keyUserProperty}_PGP'], channel=source,pageId=pageId, socialId=social_id, 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 # fb.db.reference().child(f"account/{property_id}/profile_temp/{keyUserProperty}/{valueUserProperty}").set(temp_user_pseudo_id) if 'pgpProperty' in data: for i in data['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=source,pageId=pageId, socialId=social_id, 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=source, pageId=pageId, socialId=social_id, 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() }) # for pgp in data.keys(): # if pgp.endswith('_PGP'): # if pgp in data: # data['userProperty'].append({'key': pgp, 'value': data[pgp]}) if 'pgpProperty' in data: for pgp in data['pgpProperty']: keyPGPProperty = pgp['key'] valuePGPProperty = pgp['value'] if keyPGPProperty.endswith('_PGP'): data['userProperty'].append({'key': keyPGPProperty, 'value': valuePGPProperty}) row = evntFunction.pushEventBigquery( eventId=eventId, eventTimeStamp=timeStampTransform, eventName=eventName, id=social_id, user_pseudo_id=temp_user_pseudo_id, pageId=pageId, source=source, eventProperty = data['eventProperty'] if 'eventProperty' in data else [], userProperty= data['userProperty'] if 'userProperty' in data else [], referral= data['referral'] if 'referral' in data else {}, ref_user_pseudo_id=refUser ) 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 fb.db.reference().child(f"event/{dateStr}/{hourStr}/webhook/{property_id}/{hook_id}/{eventId}").set(data) return {"status": "success", "message": "Webhook event success"}, 200 return {"status": "error", "message": "Method not allowed"}, 405