import uuid import re import hashlib import requests from datetime import datetime, timedelta from zoneinfo import ZoneInfo from dateutil import tz, parser import pandas as pd import os import pgpy import pytz import logging import json import contextlib from utility.authorization import TokenAuthorization as tAuth import utility.adminPhoneService as adminPhoneService from typing import Optional logging.basicConfig(level=logging.INFO) timezone = pytz.timezone('Asia/Bangkok') class Function: def __init__(self): self = self pass def generate_uuid_from_text(text: str, namespace_uuid: uuid.UUID = uuid.NAMESPACE_DNS) -> uuid.UUID: """ Generates a UUID (version 5) from a given text string and a namespace UUID. Args: text: The input text string to generate the UUID from. namespace_uuid: The namespace UUID. Defaults to uuid.NAMESPACE_DNS. You can use other predefined namespaces or define your own. Returns: A uuid.UUID object. """ return str(uuid.uuid5(namespace_uuid, text)) def generate_event_id(text: str, namespace_uuid: uuid.UUID = uuid.NAMESPACE_DNS) -> uuid.UUID: """ Generates a UUID (version 5) from a given text string and a namespace UUID. Args: text: The input text string to generate the UUID from. namespace_uuid: The namespace UUID. Defaults to uuid.NAMESPACE_DNS. You can use other predefined namespaces or define your own. Returns: A uuid.UUID object. """ return str(uuid.uuid5(namespace_uuid, text)) def returnPageInList(fb,accountsObject, channel=None): packPageID = [] pageDetail = [] for acc in accountsObject: fbData = fb.db.reference().child(f'property/{acc}/channel/{channel}').get(shallow=True) if fbData: for page in fbData: pageID = fb.db.reference().child(f'property/{acc}/channel/{channel}/{page}').get(shallow=True) packPageID.append(list(pageID.keys())[0]) packPage = { list(pageID.keys())[0]: { "property_id": acc, "page_order": page } } pageDetail.append(packPage) return packPageID, pageDetail def returnPageInListNew(fb, accountsObject, channel=None): packPageID = [] for acc in accountsObject: fbData = fb.db.reference().child(f'property/{acc}/channel/{channel}').get(shallow=True) if fbData: for page in fbData: packProperty = { f"{page}": { "property_id": acc } } packPageID.append(packProperty) return packPageID def returnLINEOAList(fb, accountsObject): packPageID = [] for acc in accountsObject: lineData = fb.db.reference().child(f'property/{acc}/channel/line').get(shallow=True) if lineData: for page in lineData: userId = fb.db.reference().child(f'property/{acc}/channel/line/{page}/userId').get(shallow=True) packProperty = { f"{userId}": { "property_id": acc, "channel_id": page } } packPageID.append(packProperty) return packPageID def checkPageMapping(accountMapping, id, channel): status = False accName = None for acc in accountMapping: for page in accountMapping[acc][channel]['page']: if int(accountMapping[acc][channel]['page'][page]) == int(id): status = True accName = acc return status, accName, id, page if status else None return status, None, None, None def checkAccountMapping(accountMapping, accountName=None): status = False accName = None for acc in accountMapping: if accountName and acc == accountName: status = True accName = acc return status, accName if status else None def find_phone_number(text: str): """ Finds a potential Thai phone number in the given text. It looks for numbers starting with variations of +66, 66, or 0, followed by 6, 8, or 9, and then 8 digits. The pattern accounts for common groupings and optional separators like spaces or hyphens. """ # Updated pattern explanation: # (?:(?:\+?66|0)[\s-]*)? : This is a non-capturing group for the optional prefix. # - \+?66: Matches "+66" or "66". # - |0: OR matches "0". # - [\s-]*: Matches zero or more spaces or hyphens after the prefix. # # ([689](?:[\s-]*\d){8}) : This is the main capturing group for the 9-digit phone number. # - [689]: Matches the first digit of the mobile number (must be 6, 8, or 9). # - (?:[\s-]*\d){8}: This is a non-capturing group that repeats 8 times. # - [\s-]*: Matches zero or more spaces or hyphens. # - \d: Matches a single digit. # This part ensures that the remaining 8 digits are matched, # allowing for optional spaces or hyphens between each digit. # # The entire 9-digit sequence (including internal separators) will be captured in group 1. # We then clean this captured string by removing all spaces and hyphens. pattern = r'(?:(?:\+?66|0)[\s\-]*)?([689](?:[\s\-]*\d){8})' matches = re.finditer(pattern, text) results = [] for m in matches: # Clean spaces and hyphens inside the number cleaned_number = re.sub(r'[\s\-]', '', m.group(0)) results.append(cleaned_number) return results def clean_phone_number_in_message(message: str, phone_number: str, formatted_phone: str): message = message.replace(phone_number, formatted_phone) message = message.replace(formatted_phone, phone_number[:4] + "xxx" + phone_number[6:]) return message def clean_spam_phone_number(phone_number: str): """ Cleans a potential Thai phone number and filters out common "spam" patterns. Args: phone_number: A string representing a potential phone number. Returns: The cleaned phone number string if it's not a spam pattern, otherwise None. """ if not phone_number: return None digits_only = re.sub(r'\D', '', phone_number) for i in range(10): # Check for digits 0-9 if digits_only.count(str(i)) >= 8: return None # Likely spam # Check for numbers where all 9 or 10 digits are the same (e.g., 0888888888, 0999999999) if len(set(digits_only)) == 1 and len(digits_only) >= 8: return None # Likely spam return digits_only def clean_and_format_thai_phone_number(phone_number_str: str): """ Cleans a raw phone number string and formats it to the standard Thai '66' followed by 9 digits (6xxxxxxxxx, 8xxxxxxxxx, 9xxxxxxxxx). Args: phone_number_str: The raw phone number string (e.g., "081-234-5678", "+66 81 234 5678"). Returns: The cleaned and formatted Thai phone number string (e.g., "66812345678"), or None if the input cannot be validated as a valid Thai mobile number. """ if not phone_number_str: return None cleaned_number = re.sub(r'\D', '', phone_number_str) if len(cleaned_number) == 10 and cleaned_number.startswith('0'): formatted_number = '66' + cleaned_number[1:] elif len(cleaned_number) == 9 and cleaned_number.startswith(('6', '8', '9')): formatted_number = '66' + cleaned_number elif len(cleaned_number) == 11 and cleaned_number.startswith('66'): if cleaned_number[2] in ['6', '8', '9']: formatted_number = cleaned_number else: return None elif len(cleaned_number) == 12 and cleaned_number.startswith(('0066', '+66')): temp_number = cleaned_number[-10:] if len(temp_number) == 10 and temp_number.startswith('0') and temp_number[1] in ['6', '8', '9']: formatted_number = '66' + temp_number[1:] elif len(temp_number) == 9 and temp_number[0] in ['6', '8', '9']: formatted_number = '66' + temp_number else: return None else: return None if len(formatted_number) == 11 and formatted_number.startswith('66') and formatted_number[2] in ['6', '8', '9']: return formatted_number else: return None def find_email(text: str): """ Check if the given email is valid. A valid email can contain alphanumeric characters, dots, underscores, and hyphens. """ pattern = r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}' m = re.search(pattern, text) if m: return m.group(0) return None def clean_email_in_message(message: str, email: str): # Replace the exact email with the formatted version (optional) # Mask the local part of the email local, domain = email.split("@") if len(local) <= 2: masked_email = "*" * len(local) + "@" + domain else: masked_email = local[:2] + "***@" + domain # Replace the formatted email with the masked version message = message.replace(email, masked_email) return message def hash256(text: str): """ Hash the phone number using a simple hash function. This is a placeholder for a more secure hashing mechanism. """ hash_object = hashlib.sha256(text.encode('utf-8')) return hash_object.hexdigest() def text_to_numeric_token(text, length=10): hash_object = hashlib.sha256(text.encode()) numeric_token = ''.join(filter(str.isdigit, str(int(hash_object.hexdigest(), 16)))) return int(numeric_token[:length].zfill(length)) def getBearerAccessToken(request): Authorization = request.headers.get('Authorization') bearer = Authorization.replace("Bearer ", "") return bearer def getUserID(request): Authorization = request.headers.get('Authorization') bearer = Authorization.replace("Bearer ", "") auth = tAuth(access_token=bearer) user_id = auth.get_user_id() return user_id def getUserIDFromHeader(headers): Authorization = headers.get('Authorization') bearer = Authorization.replace("Bearer ", "") auth = tAuth(access_token=bearer) user_id = auth.get_user_id() return user_id def checkRequestData(data, required_fields): """ Check if the required fields are present in the data. If any field is missing, return an error message. """ for field in required_fields: if field not in data: return {'status': 'error', 'message': f"{field} is required"}, 400 return None def format_datetime(dt_str): return datetime.fromisoformat(dt_str).strftime("%Y-%m-%d %H:%M:%S") def transformProfiletoBQ(data): result = { "createdate": datetime.fromisoformat(data["created_at"]).strftime("%Y-%m-%d %H:%M:%S"), "lastupdate": datetime.fromisoformat(data["created_at"]).strftime("%Y-%m-%d %H:%M:%S"), "user_pseudo_id": data["user_pseudo_id"], "user_profile": [] } for channel in ("facebook", "line", "instagram", "phoneNumber", "offline"): if channel in data: contexts = [] for item in data[channel].values(): contexts.append({ "id": item["id"], "created_at": datetime.fromisoformat(item["created_at"]).strftime("%Y-%m-%d %H:%M:%S"), "updated_at": datetime.fromisoformat(item["updated_at"]).strftime("%Y-%m-%d %H:%M:%S"), "source": item["source"] }) result["user_profile"].append({ "channel": channel, "context": contexts }) return result def transform_timestamp(eventTimeStamp): # Check if in milliseconds or seconds if eventTimeStamp > 10**10: ts = eventTimeStamp / 1000.0 else: ts = eventTimeStamp return datetime.fromtimestamp(ts, timezone).strftime("%Y-%m-%d %H:%M:%S") def build_order_by(sort_param: str, allowed_fields: list[str]) -> str: if not sort_param: return "" order_by_clauses = [] for item in sort_param.split(","): if "." in item: field, direction = item.split(".") direction = direction.lower() else: field, direction = item, "asc" if field not in allowed_fields or direction not in {"asc", "desc"}: continue # or raise an error order_by_clauses.append(f"`{field}` {direction.upper()}") if order_by_clauses: return "ORDER BY " + ", ".join(order_by_clauses) return "" def apply_sort(df, sort_param: str): if not sort_param: return df sort_fields = [] ascending_list = [] for item in sort_param.split(','): if '.' in item: field, direction = item.split('.') ascending = direction.lower() == 'asc' else: field = item ascending = True # default to ascending if no direction sort_fields.append(field) ascending_list.append(ascending) return df.sort_values(by=sort_fields, ascending=ascending_list, inplace=False) def to_bool(val): if isinstance(val, bool): return val if isinstance(val, str): return val.lower() in ('true', '1', 'yes') return bool(val) def prepcacheAudienceProfile(userContext): private_key = Key.load_key_from_env("private_key") finalUser = [] for user in userContext: if user != None: userData = [] for key in user: if key not in ['created_at', 'updated_at', 'user_pseudo_id']: if key.endswith("_PGP"): for log in user[key]: id = Key.pgp_decrypt(user[key][log]['id'], private_key) objectDict = { "id": id, "page_id": '-', "channel_type": '-' } userData.append(objectDict) elif key in ['line', 'facebook']: for log in user[key]: objectDict = { "id": user[key][log]['id'], "page_id": user[key][log]['source']['page_id'], "channel_type": key } userData.append(objectDict) data = { "user_pseudo_id": user['user_pseudo_id'], "profile": userData } finalUser.append(data) return finalUser def prepcacheAudienceProfileFlattened(finalUser): flattened = [] for entry in finalUser: user_id = entry['user_pseudo_id'] for p in entry['profile']: flattened.append({ 'user_pseudo_id': user_id, 'channel_type': p.get('channel_type', None), 'page_id': p.get('page_id', None), 'id': p.get('id', None) }) # Create DataFrame df = pd.DataFrame(flattened) return df def recent_by_date_only(value, time_type='day', time_num=30) -> bool: timezone = pytz.timezone('Asia/Bangkok') if time_type == 'day': cutoff_date = (datetime.now(timezone) - timedelta(days=time_num)).date() else: cutoff_date = (datetime.now(timezone) - timedelta(hours=time_num)).date() if value is None: return False if isinstance(value, str): value = parser.parse(value) if value.tzinfo is None: value = value.replace(tzinfo=timezone) else: value = value.astimezone(timezone) return value.date() >= cutoff_date def parse_mixed_to_utc(series: pd.Series) -> pd.Series: s = series.astype("string") tz_mask = s.str.contains(r'([Zz]|[+-]\d{2}:\d{2}|[+-]\d{4})$', na=False) out = pd.Series(pd.NaT, index=series.index, dtype="datetime64[ns, UTC]") aware = pd.to_datetime(s[tz_mask], errors="coerce", utc=True, format="mixed") naive = pd.to_datetime(s[~tz_mask], errors="coerce", format="mixed") naive = naive.dt.tz_localize(ZoneInfo("Asia/Bangkok")).dt.tz_convert("UTC") out.loc[tz_mask] = aware out.loc[~tz_mask] = naive return out def check_require_data(data, list_require, method): for req in list_require: if req not in data: return {"status":"error",'message': f"{req} is required"}, 400 return True def findFirebaseMaxKey(data:dict): f = list(data.keys()) max_key = max(int(k) for k in f) return int(max_key) class Event: def __init__(self): # Create an instance of ClassA directly within ClassB's constructor self.Function = Function() # You can pass arguments here if needed def socialEventName(self, eventId, pageId, socialId, channel,eventName, eventTimeStamp, eventType="standard", eventProperty=[], userProperty=[], referral={}, pgpProperty=[]): data = { 'eventId': eventId, 'pageId': pageId, 'socialId': socialId, 'channel': channel, 'eventType': eventType, 'eventName': eventName, 'eventTimeStamp': eventTimeStamp, 'eventProperty': eventProperty, 'userProperty': userProperty, 'pgpProperty': pgpProperty, 'referral': referral } return data def socialEvent(self, eventId, pageId, socialId, channel, eventName, eventTimeStamp, message=None, phoneNumber=None, referral=None, type="standard", key_unify='-'): data = { 'eventId': eventId, 'pageId': pageId, 'socialId': socialId, 'channel': channel, 'eventName': eventName, 'eventTimeStamp': eventTimeStamp, 'message': message, 'phoneNumber': phoneNumber, 'referral': referral, 'eventType': type, 'keyUnify': key_unify } return data def masterEvent(self, eventId, pageId, socialId, user_pseudo_id, channel, eventName, eventTimeStamp, message=None, phoneNumber=None, referral=None): data = { 'eventId': eventId, 'user_pseudo_id': user_pseudo_id, 'pageId': pageId, 'socialId': socialId, 'channel': channel, 'eventName': eventName, 'eventTimeStamp': eventTimeStamp, 'message': message, 'phoneNumber': phoneNumber, 'referral': referral } return data def pushEventBigquery(self, eventId, eventTimeStamp, eventName, id, user_pseudo_id, pageId, source, eventProperty: list = None, userProperty: list = None, referral: dict = None, ref_user_pseudo_id=None): data = { "eventId": eventId, "eventTimeStamp": eventTimeStamp, "eventName": eventName, "id": id, "eventProperty": eventProperty, "userProperty": userProperty, "referral": referral, "pageId": pageId, "user_pseudo_id": user_pseudo_id, "source": source, "ref_user_pseudo_id": ref_user_pseudo_id } return data def pushEventFacebookAd(self, eventId, eventTimeStamp, eventName, id, user_pseudo_id, pageId, source, ad_id): data = { "eventId": eventId, "eventTimeStamp": eventTimeStamp, "eventName": eventName, "id": id, "pageId": pageId, "user_pseudo_id": user_pseudo_id, "source": source, "ad_id": ad_id } return data def lineEventManagement(self,fb,dateStr,hourStr,parsed_data, property_id, channel_id): LINE_uid = parsed_data['events'][0]['source']['userId'] eventId = Function.generate_event_id(LINE_uid+str(datetime.now(timezone).timestamp())) eventPack = None eventType = parsed_data['events'][0]['type'] timestamp = parsed_data['events'][0]['timestamp'] # event clean process if eventType == "follow": #check follow type isUnblocked = parsed_data['events'][0]['follow']['isUnblocked'] eventName = "LINE_un_block" if isUnblocked else "LINE_add_friend" eventPack = self.socialEventName(eventId, channel_id, LINE_uid, channel="line",eventName=eventName, eventTimeStamp=timestamp) elif eventType=="unfollow": eventPack = self.socialEventName(eventId, channel_id, 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 = parsed_data['events'][0]['beacon'] beaconType = beaconObject['type'] beaconRefferal = { 'ref': beaconObject['hwid'] } beaconEventName = beaconEventMapping[beaconType] eventPack = self.socialEventName(eventId, channel_id, LINE_uid, channel="line", eventName=beaconEventName, eventTimeStamp=timestamp, referral=beaconRefferal) elif eventType == 'message': # Detect phone messageObject = parsed_data['events'][0]['message'] userMessage = messageObject["text"] phoneNumber = Function.find_phone_number(userMessage) email = Function.find_email(userMessage) #Add wrong phone number #If spam number pass if phoneNumber and email: #Phone pgpProperty = [] userProperty = [] for ph in phoneNumber: notSpam = Function.clean_spam_phone_number(ph) if not notSpam: continue env_key, admin_phones, is_exist = adminPhoneService.get_admin_phones(property_id) phoneNumberCleaned = 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 = Function.hash256(phoneNumberCleaned) messageCleaned = Function.clean_phone_number_in_message(userMessage, ph, phoneNumberCleaned) userMessage = messageCleaned pubkey = Key.load_key_from_env("public_key") phonePGP = Key.pgp_encrypt(phoneNumberCleaned, pubkey) phonePGPPack = {'key': 'phoneNumber_PGP', 'value': phonePGP} phoneHashPack = {'key': 'phoneNumber', 'value': phoneNumberHashed} pgpProperty.append(phonePGPPack) userProperty.append(phoneHashPack) #email emailHashed = Function.hash256(email) pubkey = Key.load_key_from_env("public_key") emailPGP = Key.pgp_encrypt(email, pubkey) messageCleaned = Function.clean_email_in_message(messageCleaned, email) emailPGPPack = {'key': 'email_PGP', 'value': emailPGP} emailHashPack = {'key': 'email', 'value': emailHashed} pgpProperty.append(emailPGPPack) userProperty.append(emailHashPack) # userProperty = [{'key': 'phoneNumber', 'value': phoneNumberHashed},{'key': 'email', 'value': emailHashed}] messageCleanedObject = [{'key': 'message', 'value': messageCleaned}] eventPack = self.socialEventName(eventId, pageId=channel_id, socialId=LINE_uid, channel='line', eventType='unify', eventName='contact_info_submitted', eventTimeStamp=timestamp, eventProperty=messageCleanedObject, userProperty=userProperty, pgpProperty=pgpProperty ) # eventPack.update({'phoneNumber_PGP': phonePGP}) # eventPack.update({'email_PGP': emailPGP}) elif phoneNumber: userProperty = [] pgpProperty = [] for ph in phoneNumber: env_key, admin_phones, is_exist = adminPhoneService.get_admin_phones(property_id) phoneNumberCleaned = 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 = Function.hash256(phoneNumberCleaned) messageCleaned = Function.clean_phone_number_in_message(userMessage, ph, phoneNumberCleaned) userMessage = messageCleaned pubkey = Key.load_key_from_env("public_key") phonePGP = Key.pgp_encrypt(phoneNumberCleaned, pubkey) phonePGPPack = {'key': 'phoneNumber_PGP', 'value': phonePGP} phoneHashPack = {'key': 'phoneNumber', 'value': phoneNumberHashed} pgpProperty.append(phonePGPPack) userProperty.append(phoneHashPack) # phoneObject = [{'key': 'phoneNumber', 'value': phoneNumberHashed}] messageCleanedObject = [{'key': 'message', 'value': messageCleaned}] eventPack = self.socialEventName(eventId, pageId=channel_id, socialId=LINE_uid, channel='line', eventType='unify', eventName='contact_info_submitted', eventTimeStamp=timestamp, eventProperty=messageCleanedObject, userProperty=userProperty, pgpProperty=pgpProperty ) # eventPack.update({'phoneNumber_PGP': phonePGP}) elif email: #email emailHashed = Function.hash256(email) pubkey = Key.load_key_from_env("public_key") emailPGP = Key.pgp_encrypt(email, pubkey) messageCleaned = Function.clean_email_in_message(userMessage, email) pgpProperty = [{'key': 'email_PGP', 'value': emailPGP}] userProperty = [{'key': 'email', 'value': emailHashed}] messageCleanedObject = [{'key': 'message', 'value': messageCleaned}] eventPack = self.socialEventName(eventId, pageId=channel_id, socialId=LINE_uid, channel='line', eventType='unify', eventName='contact_info_submitted', eventTimeStamp=timestamp, eventProperty=messageCleanedObject, userProperty=userProperty, pgpProperty=pgpProperty ) eventPack.update({'email_PGP': emailPGP}) else: messageCleanedObject = [{'key': 'message', 'value': userMessage}] eventPack = self.socialEventName(eventId, pageId=channel_id, socialId=LINE_uid, channel='line', eventName='user_message', eventTimeStamp=timestamp, eventProperty=messageCleanedObject, ) #Push event if eventPack: logging.info(f"\nDebug: eventPack type: {type(eventPack)}") logging.info(f"Debug: eventPack content (first 200 chars): {str(eventPack)[:200]}...") fb.db.reference().child(f"event/{dateStr}/{hourStr}/line/{property_id}/{str(channel_id)}/{eventId}").set(eventPack) return eventPack return None def facebookPageEventHandler(self, data: dict, *args, **kwargs): processed_events = [] if data.get('object') == 'page': for entry in data.get('entry', []): page_id = entry.get('id') event_time = entry.get('time') # Timestamp of the event batch for change in entry.get('changes', []): field = change.get('field') value = change.get('value') # Basic event common metadata # Initialize event_data with common fields event_data = { "page_id": page_id, "event_timestamp": event_time, "field": field, # e.g., 'feed' "hook_action": value.get('verb'), # 'add', 'remove', 'edited' "item_type": value.get('item'), # 'comment', 'reaction', 'share' "ps_id": value.get('from', {}).get('id'), "facebook_name": value.get('from', {}).get('name'), "event_name": "facecbook_unknown_event" # Default, will be updated } if field == 'feed' and value: item_type = value.get('item') verb = value.get('verb') # 'add', 'remove', 'edited' (for comments) if item_type == 'comment': comment_id = value.get('comment_id') post_id = value.get('post_id') message = value.get('message') parent_id = value.get('parent_id') # For replies event_data.update({ "event_category": "cooment_event", "comment_id": comment_id, "post_id": post_id, "comment_text": message, "is_reply": str(bool(parent_id)), "parent_comment_id": parent_id # Will be None if it's a top-level comment }) if parent_id: event_data["comment_type"] = "reply_to_comment" # Add node event name for replies event_data["event_name"] = f"comment_{verb.lower()}" else: event_data["comment_type"] = "comment_on_post" # Add node event name for top-level comments event_data["event_name"] = f"post_comment_{verb.lower()}" elif item_type == 'reaction': reaction_type = value.get('reaction_type') post_id = value.get('post_id') comment_id = value.get('comment_id') # If reaction is on a comment event_data.update({ "event_category": "reaction_event", "reaction_type": reaction_type, "target_id": comment_id if comment_id else post_id, # ID of the object reacted to "target_type": "comment" if comment_id else "post" # Type of object reacted to }) # Add node event name for reactions event_data["event_name"] = f"{reaction_type.lower()}_reaction_{verb.lower()}" elif item_type == 'share': post_id = value.get('post_id') event_data.update({ "event_category": "share_event", "shared_post_id": post_id }) # Facebook typically only sends 'add' for shares, not explicit 'remove' # Add node event name for shares event_data["event_name"] = f"share_{verb.lower()}" else: # Handle other feed items if needed, or log as unhandled event_data["event_category"] = "unhandled_feed_item" event_data["unhandled_item_data"] = value # Store the full unhandled value event_data["event_name"] = f"unhandled_feed_item_{item_type.lower()}_{verb.lower()}" processed_events.append(event_data) else: # Handle other fields if subscribed to them, or log as unhandled unhandled_event = { "page_id": page_id, "event_timestamp": event_time, "field": field, "event_category": "unhandled_field_change", "raw_value": value, # Store the raw value of the unhandled change "event_name": f"unhandled_field_change_{field.lower()}" } processed_events.append(unhandled_event) return processed_events def facebookEventManagement(self, fb, dateStr, hourStr, parsed_data, property_id, channel_id): entry = parsed_data['entry'][0] if 'messaging' in entry: messageObject = entry['messaging'][0] psid = messageObject['sender']['id'] eventId = Function.generate_event_id(psid+str(datetime.now(timezone).timestamp())) entry = parsed_data['entry'][0] eventPack = None #Check phone first if 'message' in messageObject and 'text' in messageObject['message']: userMessage = messageObject['message']['text'] phoneNumber = Function.find_phone_number(userMessage) email = Function.find_email(userMessage) #Add wrong phone number #If spam number pass # notSpam = Function.clean_spam_phone_number(phoneNumber) if phoneNumber and email: userProperty = [] pgpProperty = [] for ph in phoneNumber: notSpam = Function.clean_spam_phone_number(ph) if not notSpam: continue env_key, admin_phones, is_exist = adminPhoneService.get_admin_phones(property_id) phoneNumberCleaned = 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 = Function.hash256(phoneNumberCleaned) messageCleaned = Function.clean_phone_number_in_message(userMessage, ph, phoneNumberCleaned) userMessage = messageCleaned pubkey = Key.load_key_from_env("public_key") phonePGP = Key.pgp_encrypt(phoneNumberCleaned, pubkey) phonePGPPack = {'key': 'phoneNumber_PGP', 'value': phonePGP} phoneHashPack = {'key': 'phoneNumber', 'value': phoneNumberHashed} pgpProperty.append(phonePGPPack) userProperty.append(phoneHashPack) # phoneObject = [{'key': 'phoneNumber', 'value': phoneNumberHashed}] messageCleanedObject = [{'key': 'message', 'value': messageCleaned}] # #Phone # notSpam = Function.clean_spam_phone_number(ph) # phoneNumberCleaned = Function.clean_and_format_thai_phone_number(phoneNumber) # phoneNumberHashed = Function.hash256(phoneNumberCleaned) # messageCleaned = Function.clean_phone_number_in_message(userMessage, phoneNumber, phoneNumberCleaned) # pubkey = Key.load_key_from_env("public_key") # phonePGP = Key.pgp_encrypt(phoneNumberCleaned, pubkey) #email emailHashed = Function.hash256(email) pubkey = Key.load_key_from_env("public_key") emailPGP = Key.pgp_encrypt(email, pubkey) messageCleaned = Function.clean_email_in_message(messageCleaned, email) userMessage = messageCleaned emailPGPPack = {'key': 'email_PGP', 'value': emailPGP} emailHashPack = {'key': 'email', 'value': emailHashed} pgpProperty.append(emailPGPPack) userProperty.append(emailHashPack) # userProperty = [{'key': 'phoneNumber', 'value': phoneNumberHashed},{'key': 'email', 'value': emailHashed}] messageCleanedObject = [{'key': 'message', 'value': messageCleaned}] eventPack = self.socialEventName(eventId, pageId=channel_id, socialId=psid, channel='facebook', eventType='unify', eventName='contact_info_submitted', eventTimeStamp=entry['time'], eventProperty=messageCleanedObject, userProperty=userProperty, referral=messageObject['message']['referral'] if "referral" in messageObject['message'] else None, pgpProperty=pgpProperty ) # eventPack.update({'phoneNumber_PGP': phonePGP}) # eventPack.update({'email_PGP': emailPGP}) elif phoneNumber: userProperty = [] pgpProperty = [] for ph in phoneNumber: notSpam = Function.clean_spam_phone_number(ph) if not notSpam: continue env_key, admin_phones, is_exist = adminPhoneService.get_admin_phones(property_id) phoneNumberCleaned = 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 = Function.hash256(phoneNumberCleaned) messageCleaned = Function.clean_phone_number_in_message(userMessage, ph, phoneNumberCleaned) userMessage = messageCleaned pubkey = Key.load_key_from_env("public_key") phonePGP = Key.pgp_encrypt(phoneNumberCleaned, pubkey) phonePGPPack = {'key': 'phoneNumber_PGP', 'value': phonePGP} phoneHashPack = {'key': 'phoneNumber', 'value': phoneNumberHashed} pgpProperty.append(phonePGPPack) userProperty.append(phoneHashPack) # phoneObject = [{'key': 'phoneNumber', 'value': phoneNumberHashed}] messageCleanedObject = [{'key': 'message', 'value': messageCleaned}] # phoneNumberCleaned = Function.clean_and_format_thai_phone_number(phoneNumber) # phoneNumberHashed = Function.hash256(phoneNumberCleaned) # pubkey = Key.load_key_from_env("public_key") # phonePGP = Key.pgp_encrypt(phoneNumberCleaned, pubkey) # phoneObject = [{'key': 'phoneNumber', 'value': phoneNumberHashed}] # messageCleaned = Function.clean_phone_number_in_message(userMessage, phoneNumber, phoneNumberCleaned) # messageCleanedObject = [{'key': 'message', 'value': messageCleaned}] eventPack = self.socialEventName(eventId, pageId=channel_id, socialId=psid, channel='facebook', eventType='unify', eventName='contact_info_submitted', eventTimeStamp=entry['time'], eventProperty=messageCleanedObject, userProperty=userProperty, referral=messageObject['message']['referral'] if "referral" in messageObject['message'] else None, pgpProperty=pgpProperty ) # eventPack.update({'phoneNumber_PGP': phonePGP}) elif email: #email pgpProperty = [] userProperty = [] emailHashed = Function.hash256(email) pubkey = Key.load_key_from_env("public_key") emailPGP = Key.pgp_encrypt(email, pubkey) messageCleaned = Function.clean_email_in_message(userMessage, email) userProperty.append({'key': 'email', 'value': emailHashed}) pgpProperty.append({'email_PGP': emailPGP}) messageCleanedObject = [{'key': 'message', 'value': messageCleaned}] eventPack = self.socialEventName(eventId, pageId=channel_id, socialId=psid, channel='facebook', eventType='unify', eventName='contact_info_submitted', eventTimeStamp=entry['time'], eventProperty=messageCleanedObject, userProperty=userProperty, referral=messageObject['message']['referral'] if "referral" in messageObject['message'] else None, pgpProperty=pgpProperty ) # eventPack.update({'email_PGP': emailPGP}) else: messageCleanedObject = [{'key': 'message', 'value': userMessage}] eventPack = self.socialEventName(eventId, pageId=channel_id, 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 = self.socialEventName(eventId, pageId=channel_id, socialId=psid, channel='facebook', eventName='click_messenger_ad', eventTimeStamp=entry['time'], referral=messageObject['referral'] ) elif 'changes' in entry: changeObject = entry['changes'][0] eventPackList = self.facebookPageEventHandler(parsed_data) if eventPackList: facebookEventPack = eventPackList[0] psid = changeObject['value']['from']['id'] facebookName = changeObject['value']['from'].get("name", "-") userProperty = [{'key': 'facebook_name', 'value': facebookName}] eventId = Function.generate_event_id(psid+str(datetime.now(timezone).timestamp())) eventName = facebookEventPack['event_name'] eventTimeStamp = changeObject['value']['created_time'] facebookEventPack.pop('event_name') eventProperty = [{"key": k, "value": v} for k, v in facebookEventPack.items()] eventPack = self.socialEventName(eventId, pageId=channel_id, socialId=psid, channel='facebook',eventName=eventName, eventTimeStamp=eventTimeStamp, eventProperty=eventProperty, userProperty=userProperty) if eventPack: fb.db.reference().child(f"event/{dateStr}/{hourStr}/facebook/{property_id}/{str(channel_id)}/{eventId}").set(eventPack) return eventPack return None def facebookEventManagement_from_history_chat(self, row, property_id, channel_id, pubkey=None): """ Build a single eventPack from ONE DataFrame row (or dict) that has: thread_bucket, thread_id, unread_count, thread_updated_utc, message_id, message_text, message_created_utc, message_created_bkk, from_id, from_name, from_email, participant_ids, participant_names """ # allow both Series and dict rget = row.get if hasattr(row, "get") else lambda k, d=None: row[k] if k in row else d # cache/load pubkey once if you like (or attach to self in __init__) # try: # pubkey = Key.load_key("public_key") # except Exception: # pubkey = None timezone_bkk = tz.gettz("Asia/Bangkok") psid = str(rget("from_id", "") or "") userMessage = str(rget("message_text", "") or "") # timestamp → epoch seconds eventTimeStr = rget("message_created_utc") or rget("message_created_bkk") or "" eventTimeStamp = None if eventTimeStr: try: dt = datetime.fromisoformat(str(eventTimeStr).replace("Z", "+00:00")) eventTimeStamp = int(dt.timestamp()) except Exception: try: dt = datetime.fromisoformat(str(eventTimeStr)) if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone_bkk) eventTimeStamp = int(dt.timestamp()) except Exception: eventTimeStamp = int(datetime.now().timestamp()) else: eventTimeStamp = int(datetime.now().timestamp()) # generate event id eventTimeRaw = str(rget("message_created_bkk") or datetime.now().timestamp()) eventId = Function.generate_event_id(psid + eventTimeRaw) # detect contact info phoneNumbers = Function.find_phone_number(userMessage) email = Function.find_email(userMessage) eventPack = None if phoneNumbers and email: userProperty = [] pgpProperty = [] messageCleaned = userMessage for ph in phoneNumbers: notSpam = Function.clean_spam_phone_number(ph) if not notSpam: continue env_key, admin_phones, is_exist = adminPhoneService.get_admin_phones(property_id) phoneNumberCleaned = 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 = Function.hash256(phoneNumberCleaned) messageCleaned = Function.clean_phone_number_in_message(messageCleaned, ph, phoneNumberCleaned) if pubkey: phonePGP = Key.pgp_encrypt(phoneNumberCleaned, pubkey) pgpProperty.append({'key': 'phoneNumber_PGP', 'value': phonePGP}) userProperty.append({'key': 'phoneNumber', 'value': phoneNumberHashed}) emailHashed = Function.hash256(email) if pubkey: emailPGP = Key.pgp_encrypt(email, pubkey) pgpProperty.append({'key': 'email_PGP', 'value': emailPGP}) messageCleaned = Function.clean_email_in_message(messageCleaned, email) userProperty.append({'key': 'email', 'value': emailHashed}) messageCleanedObject = [{'key': 'message', 'value': messageCleaned}] eventPack = self.socialEventName( eventId, pageId=channel_id, socialId=psid, channel='facebook', eventType='unify', eventName='contact_info_submitted', eventTimeStamp=eventTimeStamp, eventProperty=messageCleanedObject, userProperty=userProperty, referral=None, pgpProperty=pgpProperty ) elif phoneNumbers: userProperty = [] pgpProperty = [] messageCleaned = userMessage for ph in phoneNumbers: notSpam = Function.clean_spam_phone_number(ph) if not notSpam: continue env_key, admin_phones, is_exist = adminPhoneService.get_admin_phones(property_id) phoneNumberCleaned = 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 = Function.hash256(phoneNumberCleaned) messageCleaned = Function.clean_phone_number_in_message(messageCleaned, ph, phoneNumberCleaned) if pubkey: phonePGP = Key.pgp_encrypt(phoneNumberCleaned, pubkey) pgpProperty.append({'key': 'phoneNumber_PGP', 'value': phonePGP}) userProperty.append({'key': 'phoneNumber', 'value': phoneNumberHashed}) messageCleanedObject = [{'key': 'message', 'value': messageCleaned}] eventPack = self.socialEventName( eventId, pageId=channel_id, socialId=psid, channel='facebook', eventType='unify', eventName='contact_info_submitted', eventTimeStamp=eventTimeStamp, eventProperty=messageCleanedObject, userProperty=userProperty, referral=None, pgpProperty=pgpProperty ) elif email: userProperty = [] pgpProperty = [] emailHashed = Function.hash256(email) userProperty.append({'key': 'email', 'value': emailHashed}) if pubkey: emailPGP = Key.pgp_encrypt(email, pubkey) pgpProperty.append({'key': 'email_PGP', 'value': emailPGP}) messageCleaned = Function.clean_email_in_message(userMessage, email) messageCleanedObject = [{'key': 'message', 'value': messageCleaned}] eventPack = self.socialEventName( eventId, pageId=channel_id, socialId=psid, channel='facebook', eventName='contact_info_submitted', eventTimeStamp=eventTimeStamp, eventProperty=messageCleanedObject, userProperty=userProperty, referral=None, pgpProperty=pgpProperty ) else: # no phone/email -> generic user message messageCleanedObject = [{'key': 'message', 'value': userMessage}] eventPack = self.socialEventName( eventId, pageId=channel_id, socialId=psid, channel='facebook', eventName='user_message', eventTimeStamp=eventTimeStamp, eventProperty=messageCleanedObject, referral=None ) return eventPack class Profile: def checkTelephoneNumberInProfile(profileData, telephoneNumber): """ Check if the given telephone number exists in the profile data. The profile data is expected to be a dictionary with a 'telephone' key. """ for pro in profileData: pass return False def pushNewUser(user_pseudo_id, datetime): data = { 'user_pseudo_id': user_pseudo_id, 'created_at': datetime, 'updated_at': datetime } return data def pushNewProfile(id, channel, pageId, socialId, datetime): data = { 'id': id, 'source': { "from": channel, "page_id":pageId, "id": socialId }, 'created_at': datetime, 'updated_at': datetime } return data def pushUpdateProfile(id, channel, pageId, socialId, datetime): data = { 'id': id, 'source': { "from": channel, "page_id":pageId, "id": socialId }, 'updated_at': datetime } return data def pushNewGA4Profile(id, channel, property_id, pseudo_user_id, datetime, device, geo, user_properties): data = { 'id': id, 'source': { 'from': channel, 'page_id': property_id, 'id': pseudo_user_id }, 'device': device, 'geo': geo, 'user_properties': user_properties, 'updated_at': datetime } return data def ga4UpdateSameProfile(fb, allGA4DataOld: list, temp_user_pseudo_id, pseudo_user_id, property_id, geo, device): for old in allGA4DataOld: oldPack = allGA4DataOld[old] oldGA4ID = oldPack['id'] if oldGA4ID == pseudo_user_id: #Update fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/ga4/{old}/geo").set(geo) fb.db.reference().child(f"account/{property_id}/profile/{temp_user_pseudo_id}/ga4/{old}/device").set(device) class Notify: def __init__(self, key=os.environ.get("GOOGLE_CHAT_NOTI_KEY")): self.webhook_url = f"https://chat.googleapis.com/v1/spaces/AAQAYTm0RE8/messages?key={key}" def googleChatNotify(self, message): """ Send a notification to Google Chat. This function is a placeholder and should be implemented with actual logic. """ headers = { 'Content-Type': 'application/json', } data = { "text": message } response = requests.post(self.webhook_url, headers=headers, json=data) if response.status_code == 200: return True else: return False class Pack: def packOverviewDashboard(totalUsers, line1_userSeries, totalEvent, line1_eventSeries, line1_xaxis, totalMessage, line2_MessageSeries, messageLList, ads_Click, ads_Message): data = { "total_user": { "title": "Total Users", "count": f"{totalUsers:.0f}", "series": line1_userSeries }, "total_event": { "title": "Total Events", "count": f"{totalEvent:.0f}", "series": line1_eventSeries }, "event_user_traffic": { "series": [ { "name": "Users", "type": "area", "data": line1_userSeries }, { "name": "Events", "type": "area", "data": line1_eventSeries } ], "xaxis": { "categories": line1_xaxis } }, "example_messages": { "total_message": { "title": "Total Messages", "count": f"{totalMessage:.0f}", "series": line2_MessageSeries }, "messages": messageLList }, "ads_engagement": { "series": [ { "name": "Click Messenger Ads", "type": "area", "data": ads_Click }, { "name": "Message within Ads", "type": "area", "data": ads_Message } ], "xaxis": { "categories": line1_xaxis } } } return data class KeyPGP: def __init__(self): pass def load_private_key_local(self, filepath: str, passphrase: str = None) -> pgpy.PGPKey: key, _ = pgpy.PGPKey.from_file(filepath) if key.is_protected and passphrase: key.unlock(passphrase) return key def load_key_from_env(self, env_var_name: str, passphrase: str = None) -> pgpy.PGPKey: value = os.environ.get(env_var_name) if not value: raise ValueError(f"Environment variable {env_var_name} is not set.") # If env contains a path to a .txt file → load from file if value.lower().endswith(".txt"): return self.load_private_key_local(value, passphrase) # Otherwise treat env as the key blob key_blob = value.replace('\\n', '\n') key, _ = pgpy.PGPKey.from_blob(key_blob) if key.is_protected and passphrase: key.unlock(passphrase) return key def pgp_decrypt(self, encrypted_text: str, privkey: pgpy.PGPKey, passphrase: str = None) -> str: message = pgpy.PGPMessage.from_blob(encrypted_text) if privkey.is_protected: if passphrase is None: raise ValueError("Passphrase required for protected private key.") with privkey.unlock(passphrase): decrypted = privkey.decrypt(message) else: decrypted = privkey.decrypt(message) return str(decrypted.message) class Key: # Load keys def load_private_key_local(self, filepath: str, passphrase: str = None) -> pgpy.PGPKey: key, _ = pgpy.PGPKey.from_file(filepath) if key.is_protected and passphrase: key.unlock(passphrase) return key # def load_key_from_env(env_var_name: str, passphrase: str = None) -> pgpy.PGPKey: # key_blob = os.environ.get(env_var_name) # if not key_blob: # raise ValueError(f"Environment variable {env_var_name} is not set.") # key_blob = key_blob.replace('\\n', '\n') # key, _ = pgpy.PGPKey.from_blob(key_blob) # if key.is_protected and passphrase: # key.unlock(passphrase) # return key def load_key_from_env(self, env_var_name: str, passphrase: str = None) -> pgpy.PGPKey: value = os.environ.get(env_var_name) if not value: raise ValueError(f"Environment variable {env_var_name} is not set.") # If env contains a path to a .txt file → load from file if value.lower().endswith(".txt"): return self.load_private_key_local(value, passphrase) # Otherwise treat env as the key blob key_blob = value.replace('\\n', '\n') key, _ = pgpy.PGPKey.from_blob(key_blob) if key.is_protected and passphrase: key.unlock(passphrase) return key # Encryption/Decryption def pgp_encrypt(text: str, pubkey: pgpy.PGPKey) -> str: message = pgpy.PGPMessage.new(text) encrypted_message = pubkey.encrypt(message) return str(encrypted_message) def pgp_decrypt(encrypted_text: str, privkey: pgpy.PGPKey, passphrase: str = None) -> str: message = pgpy.PGPMessage.from_blob(encrypted_text) if privkey.is_protected: if passphrase is None: raise ValueError("Passphrase required for protected private key.") with privkey.unlock(passphrase): decrypted = privkey.decrypt(message) else: decrypted = privkey.decrypt(message) return str(decrypted.message) SHA256_HEX_RE = re.compile(r'^[A-Fa-f0-9]{64}$') SHA256_TAGGED_RE = re.compile(r'^(sha256:)?[A-Fa-f0-9]{64}$', re.IGNORECASE) PGP_ARMOR_HEADER = "-----BEGIN PGP MESSAGE-----" def looks_sha256_hex(v: str) -> bool: if not isinstance(v, str): return False s = v.strip() # Support either pure 64-hex or optional "sha256:" prefix return bool(Key.SHA256_TAGGED_RE.fullmatch(s)) def looks_pgp_encrypted(v: str) -> bool: return isinstance(v, str) and v.lstrip().startswith(Key.PGP_ARMOR_HEADER) def encrypt_dataframe_columns(df: pd.DataFrame, columns: list, pubkey: pgpy.PGPKey): """Encrypt columns (no null/hash/PGP checks).""" out = df.copy() for col in columns: if col in out.columns: out[col] = out[col].apply(lambda x: Key.pgp_encrypt(str(x), pubkey)) return out def encrypt_dataframe_columns_ignore_null(df: pd.DataFrame, columns: list, pubkey: pgpy.PGPKey): """ Encrypt only when: - value is not null/blank, - value is not already PGP-encrypted, - value is not already a SHA-256 hash (hex, optionally prefixed 'sha256:'). """ def _encrypt_if_needed(v): # Skip nulls (NaN/None/NaT) and empty/whitespace-only strings if pd.isna(v) or (isinstance(v, str) and v.strip() == ""): return v # Skip if already PGP armored if Key.looks_pgp_encrypted(v): return v # Skip if already hashed (sha256 hex) if Key.looks_sha256_hex(v): return v # Otherwise encrypt return Key.pgp_encrypt(str(v), pubkey) out = df.copy() for col in columns: if col in out.columns: out[col] = out[col].apply(_encrypt_if_needed) return out def is_pgp_armored(value) -> bool: """Lightweight check for ASCII-armored PGP content.""" PGP_ARMOR_HEADER = "-----BEGIN PGP MESSAGE-----" return isinstance(value, str) and value.lstrip().startswith(PGP_ARMOR_HEADER) def decrypt_dataframe_columns_ignore_null( df: pd.DataFrame, columns: list, privkey: pgpy.PGPKey, passphrase: Optional[str] = None, errors: str = "keep", # "keep" | "set_na" | "raise" ) -> pd.DataFrame: """ Decrypt selected columns, skipping null/empty values and non-PGP text. Parameters ---------- df : DataFrame Source data. columns : list[str] Column names to decrypt. privkey : pgpy.PGPKey Private key used for decryption. passphrase : str | None Passphrase to unlock the private key, if protected. errors : {"keep","set_na","raise"} - "keep": leave the original value on decryption failure (default) - "set_na": set pd.NA on failure - "raise": re-raise the exception Returns ------- DataFrame A copy of `df` with requested columns decrypted where applicable. """ def _handle_error(v, exc): if errors == "keep": return v if errors == "set_na": return pd.NA raise exc def _decrypt_if_needed(v): # Skip nulls and empty/whitespace strings if pd.isna(v) or (isinstance(v, str) and v.strip() == ""): return v # If it doesn't look like PGP armor, leave it alone if not Key.is_pgp_armored(v): return v # Attempt decryption try: msg = pgpy.PGPMessage.from_blob(v) ctx = privkey.unlock(passphrase) if passphrase else contextlib.nullcontext(privkey) with ctx: dec = privkey.decrypt(msg) return dec.message # str plaintext except Exception as e: return _handle_error(v, e) out = df.copy() for col in columns: if col in out.columns: out[col] = out[col].apply(_decrypt_if_needed) else: # silently skip missing columns; swap to KeyError if you prefer continue return out # Decrypt 2 specific columns def decrypt_dataframe_columns(df: pd.DataFrame, encrypted_cols: list, privkey: pgpy.PGPKey, passphrase: str = None): for col in encrypted_cols: df[col] = df[col].apply(lambda x: Key.pgp_decrypt(x, privkey, passphrase)) return df class Unify: def checkUserIdExist(fb, property_id, user_id): val = fb.db.reference().child(f"account/{property_id}/profile/{user_id}").get(shallow=True) return val if val else None def checkMapping(fb, property_id, seaarch): val = fb.db.reference().child(f"account/{property_id}/mapping/{seaarch}").get(shallow=True) return val if val else None def checkUserPropertyTemp(fb, property_id, search_type, search): propertyProfile = fb.db.reference().child(f"account/{property_id}/profile_temp/{search_type}/{search}").get(shallow=True) return propertyProfile if propertyProfile else None def checkAllbyType(fb, property_id, search_type, search): propertyProfile = fb.db.reference().child(f"account/{property_id}/profile").get(shallow=True) for pro in propertyProfile: profileKey = fb.db.reference().child(f"account/{property_id}/profile/{pro}").get(shallow=True) if search_type not in profileKey: continue else: userProfile = fb.db.reference().child(f"account/{property_id}/profile/{pro}/{search_type}").get(shallow=True) for log in userProfile: id = fb.db.reference().child(f"account/{property_id}/profile/{search_type}/id/{log}/id").get(shallow=True) if str(id) != str(search): continue else: return pro return None def checkPropertyIdByChannelId(fb, channel_type, channel_id): propertyList = fb.db.reference().child(f"property").get(shallow=True) for pro in propertyList: channel = fb.db.reference().child(f"property/{pro}/channel/{channel_type}/{channel_id}").get(shallow=True) if not channel: continue else: return pro def checkUserProperty(fb, property_id, user_id, channel_type): profileKey = fb.db.reference().child(f"account/{property_id}/profile/{user_id}/{channel_type}").get(shallow=True) if profileKey: return True return False def insertMapping(fb, property_id, key, value): fb.db.reference().child(f"account/{property_id}/mapping/{key}").set(value) def addProfileProperty(fb, property_id, user_id, context): fb.db.reference().child(f"account/{property_id}/profile/{user_id}").set(context) def checkParentProfile(fb, property_id, user_id): allMapping = fb.db.reference().child(f"account/{property_id}/mapping").get() if allMapping: for i in allMapping: if i == user_id: return allMapping[i] return None return None