from flask import Blueprint, request, Response import logging import uuid import pytz import json import pandas as pd import numpy as np from google.cloud import bigquery from firebase.firebase import Firebase from collections import Counter from datetime import datetime from bigquery.bq import BigQuery, Schema from utility.function import Function, Pack, Key, Profile from utility.offline import Offline from utility.token import Token from utility.authorization import TokenAuthorization logging.basicConfig(level=logging.INFO) schema_data = Blueprint('schema_data', __name__, url_prefix='/data') fb = Firebase(host="https://ydm-profile-360-default-rtdb.asia-southeast1.firebasedatabase.app/") timezone = pytz.timezone('Asia/Bangkok') @schema_data.route('/insert_data', methods=['GET','POST','DELETE']) def insert_data(): if request.method == 'POST': data = request.get_json() for req in ['property_id','table_name','data']: if req not in data: return {"status":"error",'message': f"{req} is require for inserting data"}, 400 property_id = data.get('property_id', None) table_name = data.get('table_name', None) insert_data = data.get('data', None) if not isinstance(insert_data, list): return {"status":"error",'message': "Data must be a list of records"}, 400 df = pd.DataFrame(data.get("data",[])) demographic = fb.db.reference(f'property/{property_id}/offline_mapping/offline_{table_name}/demographic_col').get() contact_key = fb.db.reference(f'property/{property_id}/offline_mapping/offline_{table_name}/contact_key').get() other_key = fb.db.reference(f'property/{property_id}/offline_mapping/offline_{table_name}/other_key').get() event_mapping = fb.db.reference(f'property/{property_id}/offline_mapping/offline_{table_name}/event_mapping').get() df['version_key'] = datetime.now().strftime('%Y-%m-%dT%H:%M:%SZ') public_key = Key.load_key_from_env('D:/public_key_360.txt') sensitive_col = fb.db.reference(f'property/{property_id}/offline_mapping/offline_{table_name}/sensitive_data').get() key_columns = list(fb.db.reference(f'property/{property_id}/offline_mapping/offline_{table_name}/contact_key').get().values()) remaining_sensitive = [x for x in sensitive_col if x not in key_columns] if 'user_pseudo_id' not in df.columns: df['user_pseudo_id'] = None ## ------ unification part ------ ## # identify unify column (telephone or email) phone_col = fb.db.reference(f'property/{property_id}/offline_mapping/offline_{table_name}/contact_key/Phone').get() email_col = fb.db.reference(f'property/{property_id}/offline_mapping/offline_{table_name}/contact_key/Email').get() customer_id_col = fb.db.reference(f'property/{property_id}/offline_mapping/offline_{table_name}/other_key/CustomerID').get() # Has Phone column if phone_col: profile_temp_phone = fb.db.reference(f'account/{property_id}/profile_temp/phoneNumber').get() or {} profile_temp_email = fb.db.reference(f'account/{property_id}/profile_temp/email').get() or {} profile_temp_customer_id = fb.db.reference(f'account/{property_id}/profile_temp/customer_id').get() or {} df[phone_col] = df[phone_col].astype(str).str.replace('.0','') for idx, row in df.iterrows(): phone = row[phone_col] if email_col: email = row[email_col] else: email = None version_key = row["version_key"] print(row.name,phone, email, version_key) # each record contain Phone if pd.notna(phone) and phone != "" and phone.lower() != "nan": hash_phone = Offline.normalize_phone(str(phone)) # print("Hash Phone:", hash_phone) # check for pseudoId for this phone if hash_phone in profile_temp_phone: phone_psuedo = profile_temp_phone[hash_phone] # print("Matched by phone:", phone_psuedo) # each record contain Email if pd.notna(email): # print("Email:", email) hash_email = Offline.normalize_email(str(email)) # check for pseudoId for this email if hash_email in profile_temp_email: # print("Matched by email:", profile_temp_email[hash_email]) if profile_temp_email[hash_email] == profile_temp_phone[hash_phone]: # print("SCENARIO 1:") # print("Matched by phone and email: Adding other information") if customer_id_col and row[customer_id_col] in profile_temp_customer_id: old_pseudo = profile_temp_customer_id[row[customer_id_col]] fb.db.reference(f'account/{property_id}/profile_temp/CustomerID/{row[customer_id_col]}').set(phone_psuedo) Offline.log_change_customer_pseudo(property_id, table_name, version_key, old_pseudo, phone_psuedo, "CustomerID", row[customer_id_col]) Offline.log_mapped_fields(property_id, table_name, version_key, phone_psuedo, row, other_key, remaining_sensitive, public_key) Offline.log_mapped_fields(property_id, table_name, version_key, phone_psuedo, row, demographic, remaining_sensitive, public_key) df.at[idx, 'user_pseudo_id'] = phone_psuedo else: # print("SCENARIO 2:") # print("Psuedo from phone and email are different, merge them") # print("Change email temp from ", profile_temp_email[hash_email], " to ", profile_temp_phone[hash_phone]) fb.db.reference(f'account/{property_id}/profile_temp/email/{hash_email}').set(profile_temp_phone[hash_phone]) fb.db.reference(f'account/{property_id}/mapping/{profile_temp_email[hash_email]}').set(profile_temp_phone[hash_phone]) Offline.log_merge(property_id, table_name, version_key, profile_temp_email[hash_email], profile_temp_phone[hash_phone], "email", hash_email) Offline.copy_profile_with_logging(property_id, table_name,version_key,old_pseudo=profile_temp_email[hash_email], new_pseudo=profile_temp_phone[hash_phone]) Offline.log_mapped_fields(property_id, table_name, version_key, phone_psuedo, row, other_key, remaining_sensitive, public_key) Offline.log_mapped_fields(property_id, table_name, version_key, phone_psuedo, row, demographic, remaining_sensitive, public_key) if customer_id_col and row[customer_id_col] in profile_temp_customer_id: old_pseudo = profile_temp_customer_id[row[customer_id_col]] fb.db.reference(f'account/{property_id}/profile_temp/CustomerID/{row[customer_id_col]}').set(phone_psuedo) Offline.log_change_customer_pseudo(property_id, table_name, version_key, old_pseudo, phone_psuedo, "CustomerID", row[customer_id_col]) df.at[idx, 'user_pseudo_id'] = phone_psuedo # no psuedoId for this email else: # print("SCENARIO 3:") # print("No psudoID for this email, Adding email and other detail to psudoID: ", phone_psuedo) # check if email is hashed if Offline.is_sha256_hex(str(email)): Offline.add_details_and_log(property_id, table_name, version_key, phone_psuedo, hash_email, "email") else: Offline.add_details_and_log(property_id, table_name, version_key, phone_psuedo, hash_email, "email") Offline.add_details_and_log(property_id, table_name, version_key, phone_psuedo, Key.pgp_encrypt(email,public_key), "email_PGP") fb.db.reference(f'account/{property_id}/profile_temp/email/{hash_email}').set(phone_psuedo) if customer_id_col and row[customer_id_col] in profile_temp_customer_id: old_pseudo = profile_temp_customer_id[row[customer_id_col]] fb.db.reference(f'account/{property_id}/profile_temp/CustomerID/{row[customer_id_col]}').set(phone_psuedo) Offline.log_change_customer_pseudo(property_id, table_name, version_key, old_pseudo, phone_psuedo, "CustomerID", row[customer_id_col]) Offline.log_mapped_fields(property_id, table_name, version_key, phone_psuedo, row, other_key, remaining_sensitive, public_key) Offline.log_mapped_fields(property_id, table_name, version_key, phone_psuedo, row, demographic, remaining_sensitive, public_key) df.at[idx, 'user_pseudo_id'] = phone_psuedo # email is null else: phone_psuedo = profile_temp_phone[hash_phone] # print("SCENARIO 4:") # print("Email is null, Add only other information to Profile") Offline.log_mapped_fields(property_id, table_name, version_key, phone_psuedo, row, other_key, remaining_sensitive, public_key) Offline.log_mapped_fields(property_id, table_name, version_key, phone_psuedo, row, demographic, remaining_sensitive, public_key) if customer_id_col and row[customer_id_col] in profile_temp_customer_id: old_pseudo = profile_temp_customer_id[row[customer_id_col]] fb.db.reference(f'account/{property_id}/profile_temp/CustomerID/{row[customer_id_col]}').set(phone_psuedo) Offline.log_change_customer_pseudo(property_id, table_name, version_key, old_pseudo, phone_psuedo, "CustomerID", row[customer_id_col]) df.at[idx, 'user_pseudo_id'] = phone_psuedo # no psuedoId for this phone else: # print("No psudoID for this phone") if pd.notna(email): # print("Email:", email) hash_email = Offline.normalize_email(str(email)) # check for pseudoId for this email if hash_email in profile_temp_email: # print("SCENARIO 5:") # print("Matched by email:", profile_temp_email[hash_email]) # check if phone is hashed if Offline.is_sha256_hex(str(phone)): Offline.add_details_and_log(property_id, table_name, version_key, profile_temp_email[hash_email], hash_phone, "phoneNumber") else: Offline.add_details_and_log(property_id, table_name, version_key, profile_temp_email[hash_email], hash_phone, "phoneNumber") Offline.add_details_and_log(property_id, table_name, version_key, profile_temp_email[hash_email], Key.pgp_encrypt(phone,public_key), "phoneNumber_PGP") fb.db.reference(f'account/{property_id}/profile_temp/phoneNumber/{hash_phone}').set(profile_temp_email[hash_email]) Offline.log_mapped_fields(property_id, table_name, version_key, profile_temp_email[hash_email], row, other_key, remaining_sensitive, public_key) Offline.log_mapped_fields(property_id, table_name, version_key, profile_temp_email[hash_email], row, demographic, remaining_sensitive, public_key) if customer_id_col and row[customer_id_col] in profile_temp_customer_id: old_pseudo = profile_temp_customer_id[row[customer_id_col]] fb.db.reference(f'account/{property_id}/profile_temp/CustomerID/{row[customer_id_col]}').set(profile_temp_email[hash_email]) Offline.log_change_customer_pseudo(property_id, table_name, version_key, old_pseudo, profile_temp_email[hash_email], "CustomerID", row[customer_id_col]) df.at[idx, 'user_pseudo_id'] = profile_temp_email[hash_email] # no psuedoId for this email else: # print("SCENARIO 6:") # print("No psudoID for this email") new_pseudo = Function.generate_uuid_from_text(hash_phone) # print("Generate psuedo_id using phoneNumber ", new_pseudo) Offline.log_gen_pseudo(property_id, table_name, version_key, 'phoneNumber', hash_phone, new_pseudo) fb.db.reference(f'account/{property_id}/profile_temp/phoneNumber/{hash_phone}').set(new_pseudo) fb.db.reference(f'account/{property_id}/profile_temp/email/{hash_email}').set(new_pseudo) if Offline.is_sha256_hex(str(phone)) == False: Offline.add_details_and_log(property_id, table_name, version_key, new_pseudo, Key.pgp_encrypt(phone,public_key), "phoneNumber_PGP") if Offline.is_sha256_hex(str(email)) == False: Offline.add_details_and_log(property_id, table_name, version_key, new_pseudo, Key.pgp_encrypt(email,public_key), "email_PGP") Offline.log_mapped_fields(property_id, table_name, version_key, new_pseudo, row, other_key, remaining_sensitive, public_key) Offline.log_mapped_fields(property_id, table_name, version_key, new_pseudo, row, demographic, remaining_sensitive, public_key) if customer_id_col and row[customer_id_col] in profile_temp_customer_id: old_pseudo = profile_temp_customer_id[row[customer_id_col]] fb.db.reference(f'account/{property_id}/profile_temp/CustomerID/{row[customer_id_col]}').set(new_pseudo) Offline.log_change_customer_pseudo(property_id, table_name, version_key, old_pseudo, new_pseudo, "CustomerID", row[customer_id_col]) df.at[idx, 'user_pseudo_id'] = new_pseudo else: new_pseudo = Function.generate_uuid_from_text(hash_phone) # print("SCENARIO 7:") # print("Email is null, Generate new pseudoID for this phone and add other info: ", new_pseudo) Offline.log_gen_pseudo(property_id, table_name, version_key, 'phoneNumber', hash_phone, new_pseudo) fb.db.reference(f'account/{property_id}/profile_temp/phoneNumber/{hash_phone}').set(new_pseudo) if Offline.is_sha256_hex(str(phone)) == False: Offline.add_details_and_log(property_id, table_name, version_key, new_pseudo, Key.pgp_encrypt(phone,public_key), "phoneNumber_PGP") Offline.log_mapped_fields(property_id, table_name, version_key, new_pseudo, row, other_key, remaining_sensitive, public_key) Offline.log_mapped_fields(property_id, table_name, version_key, new_pseudo, row, demographic, remaining_sensitive, public_key) if customer_id_col and row[customer_id_col] in profile_temp_customer_id: old_pseudo = profile_temp_customer_id[row[customer_id_col]] fb.db.reference(f'account/{property_id}/profile_temp/CustomerID/{row[customer_id_col]}').set(new_pseudo) Offline.log_change_customer_pseudo(property_id, table_name, version_key, old_pseudo, new_pseudo, "CustomerID", row[customer_id_col]) df.at[idx, 'user_pseudo_id'] = new_pseudo # phone is null else: # print("Phone is null") # email exist if pd.notna(email): # print("Email:", email) hash_email = Offline.normalize_email(str(email)) # check for pseudoId for this email if hash_email in profile_temp_email: # print("SCENARIO 8:") # print("Matched by email:", profile_temp_email[hash_email], "Add other information to Profile") Offline.log_mapped_fields(property_id, table_name, version_key, profile_temp_email[hash_email], row, other_key, remaining_sensitive, public_key) Offline.log_mapped_fields(property_id, table_name, version_key, profile_temp_email[hash_email], row, demographic, remaining_sensitive, public_key) if customer_id_col and row[customer_id_col] in profile_temp_customer_id: old_pseudo = profile_temp_customer_id[row[customer_id_col]] fb.db.reference(f'account/{property_id}/profile_temp/CustomerID/{row[customer_id_col]}').set(profile_temp_email[hash_email]) Offline.log_change_customer_pseudo(property_id, table_name, version_key, old_pseudo, profile_temp_email[hash_email], "CustomerID", row[customer_id_col]) df.at[idx, 'user_pseudo_id'] = profile_temp_email[hash_email] # no psuedoId for this email else: # print("SCENARIO 9:") # print("No psudoID for this email") new_pseudo = Function.generate_uuid_from_text(hash_email) # print("Generate psuedo_id using email ", new_pseudo) Offline.log_gen_pseudo(property_id, table_name, version_key, 'email', hash_email, new_pseudo) fb.db.reference(f'account/{property_id}/profile_temp/email/{hash_email}').set(new_pseudo) if Offline.is_sha256_hex(str(email)) == False: Offline.add_details_and_log(property_id, table_name, version_key, new_pseudo, Key.pgp_encrypt(email,public_key), "email_PGP") Offline.log_mapped_fields(property_id, table_name, version_key, new_pseudo, row, other_key, remaining_sensitive, public_key) Offline.log_mapped_fields(property_id, table_name, version_key, new_pseudo, row, demographic, remaining_sensitive, public_key) if customer_id_col and row[customer_id_col] in profile_temp_customer_id: old_pseudo = profile_temp_customer_id[row[customer_id_col]] fb.db.reference(f'account/{property_id}/profile_temp/CustomerID/{row[customer_id_col]}').set(new_pseudo) Offline.log_change_customer_pseudo(property_id, table_name, version_key, old_pseudo, new_pseudo, "CustomerID", row[customer_id_col]) df.at[idx, 'user_pseudo_id'] = new_pseudo else: # print("SCENARIO 10:") # print("Email is null And Phone is null") if customer_id_col in list(other_key.values()): new_pseudo = Function.generate_uuid_from_text(row[customer_id_col]) # print(f"Generate psuedo_id using customer_id {row[customer_id_col]}: ", new_pseudo) Offline.log_gen_pseudo(property_id, table_name, version_key, 'CustomerID', row[customer_id_col], new_pseudo) fb.db.reference(f'account/{property_id}/profile_temp/CustomerID/{row[customer_id_col]}').set(new_pseudo) Offline.log_mapped_fields(property_id, table_name, version_key, new_pseudo, row, other_key, remaining_sensitive, public_key) Offline.log_mapped_fields(property_id, table_name, version_key, new_pseudo, row, demographic, remaining_sensitive, public_key) df.at[idx, 'user_pseudo_id'] = new_pseudo else: continue # print("There is not enough key to generate new profile. Skipping...") print() elif email_col: # print(f"This file has no Phone, but has email column named: {email_col}") profile_temp_email = fb.db.reference(f'account/{property_id}/profile_temp/email').get() or {} profile_temp_customer_id = fb.db.reference(f'account/{property_id}/profile_temp/customer_id').get() or {} for idx, row in df.iterrows(): email = row[email_col] if customer_id_col: customer_id = row[customer_id_col] else: customer_id = None version_key = row["version_key"] # each record contain Email if pd.notna(email) and email != "" and email.lower() != "nan": hash_email = Offline.normalize_email(str(email)) # print("Hash Email:", hash_email) # check for pseudoId for this email if hash_email in profile_temp_email: # print("SCENARIO 11:") # print("Matched by email:", profile_temp_email[hash_email], "Add other information to Profile") Offline.log_mapped_fields(property_id, table_name, version_key, profile_temp_email[hash_email], row, other_key, remaining_sensitive, public_key) Offline.log_mapped_fields(property_id, table_name, version_key, profile_temp_email[hash_email], row, demographic, remaining_sensitive, public_key) if customer_id_col and row[customer_id_col] in profile_temp_customer_id: old_pseudo = profile_temp_customer_id[row[customer_id_col]] fb.db.reference(f'account/{property_id}/profile_temp/customer_id/{row[customer_id_col]}').set(profile_temp_email[hash_email]) Offline.log_change_customer_pseudo(property_id, table_name, version_key, old_pseudo, profile_temp_email[hash_email], "CustomerID", row[customer_id_col]) df.at[idx, 'user_pseudo_id'] = profile_temp_email[hash_email] # no psuedoId for this email else: # print("SCENARIO 12:") # print("No psudoID for this email") new_pseudo = Function.generate_uuid_from_text(hash_email) # print("Generate psuedo_id using email ", new_pseudo) Offline.log_gen_pseudo(property_id, table_name, version_key, 'email', hash_email, new_pseudo) fb.db.reference(f'account/{property_id}/profile_temp/email/{hash_email}').set(new_pseudo) if Offline.is_sha256_hex(str(email)) == False: Offline.add_details_and_log(property_id, table_name, version_key, new_pseudo, Key.pgp_encrypt(email,public_key), "email_PGP") Offline.log_mapped_fields(property_id, table_name, version_key, new_pseudo, row, other_key, remaining_sensitive, public_key) Offline.log_mapped_fields(property_id, table_name, version_key, new_pseudo, row, demographic, remaining_sensitive, public_key) if customer_id_col and row[customer_id_col] in profile_temp_customer_id: old_pseudo = profile_temp_customer_id[row[customer_id_col]] fb.db.reference(f'account/{property_id}/profile_temp/customer_id/{row[customer_id_col]}').set(new_pseudo) Offline.log_change_customer_pseudo(property_id, table_name, version_key, old_pseudo, new_pseudo, "CustomerID", row[customer_id_col]) df.at[idx, 'user_pseudo_id'] = new_pseudo else: # print("SCENARIO 13:") # print("Email is null and No Phone Column") if customer_id_col in list(other_key.values()): new_pseudo = Function.generate_uuid_from_text(row[customer_id_col]) # print(f"Generate psuedo_id using customer_id {row[customer_id_col]}: ", new_pseudo) Offline.log_gen_pseudo(property_id, table_name, version_key, 'CustomerID', row[customer_id_col], new_pseudo) fb.db.reference(f'account/{property_id}/profile_temp/CustomerID/{row[customer_id_col]}').set(new_pseudo) Offline.log_mapped_fields(property_id, table_name, version_key, new_pseudo, row, other_key, remaining_sensitive, public_key) Offline.log_mapped_fields(property_id, table_name, version_key, new_pseudo, row, demographic, remaining_sensitive, public_key) df.at[idx, 'user_pseudo_id'] = new_pseudo else: # print("There is not enough key to generate new profile. Skipping...") continue print() elif customer_id_col: for idx, row in df.iterrows(): # print("SCENARIO 14:") email = row[email_col] version_key = row["version_key"] new_pseudo = Function.generate_uuid_from_text(row[customer_id_col]) # print(f"Generate psuedo_id using customer_id {row[customer_id_col]}: ", new_pseudo) Offline.log_gen_pseudo(property_id, table_name, version_key, 'CustomerID', row[customer_id_col], new_pseudo) fb.db.reference(f'account/{property_id}/profile_temp/CustomerID/{row[customer_id_col]}').set(new_pseudo) Offline.log_mapped_fields(property_id, table_name, version_key, new_pseudo, row, other_key, remaining_sensitive, public_key) Offline.log_mapped_fields(property_id, table_name, version_key, new_pseudo, row, demographic, remaining_sensitive, public_key) df.at[idx, 'user_pseudo_id'] = new_pseudo unique_col = fb.db.reference(f'property/{property_id}/offline_mapping/offline_{table_name}/unique_col').get() on_clause = Offline.build_on(unique_col) df_encrypted = Key.encrypt_dataframe_columns(df, sensitive_col, public_key) df_encrypted['version_key'] = pd.to_datetime(df_encrypted['version_key']) bq = BigQuery() bq.delete_data("customer-360-profile",f"client_{property_id}",f"offline_{table_name}_temp") bq.load_data_df(f"client_{property_id}",f"offline_{table_name}_temp",df_encrypted) bq.delete_when_match("customer-360-profile",f"client_{property_id}",f"offline_{table_name}", f"client_{property_id}",f"offline_{table_name}_temp", f"ON {on_clause} ") bq.load_data_df(f'client_{property_id}',f"offline_{table_name}",df_encrypted) count_row = bq.bq_count_rows(f"offline_{table_name}") fb.db.reference(f'property/{property_id}/offline_mapping/offline_{table_name}/row_count').set(count_row) return {"status": "Data inserted successfully"}, 200 return {"status":"error",'message': "Method not allowed"}, 405 @schema_data.route('/truncate_table', methods=['GET','POST','DELETE']) def truncate_table(): if request.method == "POST": data = request.get_json() for req in ['property_id','table_name']: if req not in data: return {"status":"error",'message': f"{req} is require for truncating table"}, 400 property_id = data.get('property_id', None) table_name = data.get('table_name', None) bq = BigQuery() bq.delete_data("customer-360-profile",f"client_{property_id}",f"offline_{table_name}") return {"status": f"Table '{table_name}' truncated successfully"}, 200 @schema_data.route('/delete_data', methods=['GET','POST','DELETE']) def delete_data_using_version_key(): if request.method == "POST": data = request.get_json() for req in ['property_id','table_name','version_key']: if req not in data: return {"status":"error",'message': f"{req} is require for deleting data"}, 400 property_id = data.get('property_id', None) table_name = data.get('table_name', None) version_key = data.get('version_key', None) log_details = fb.db.reference(f'account/{property_id}/offline_log/{table_name}/{version_key}').get() ## Remove all pseudo that was created from this version_key for each_pseudo, change_detail in log_details['gen_pseudo'].items(): fb.db.reference(f'account/{property_id}/profile/{each_pseudo}').delete() print(f"Deleting {each_pseudo}") for key, val in change_detail.items(): fb.db.reference(f'account/{property_id}/profile_temp/{key}/{val["id"]}').delete() # Remove additional details that were added by this version_key for each_pseudo in list(log_details['add_details'].keys()): for each_key in list(log_details['add_details'][each_pseudo].keys()): for key, each_detail in list(log_details['add_details'][each_pseudo][each_key].items()): if (fb.db.reference(f'account/{property_id}/profile/{each_pseudo}/{each_key}').get()): for profile_key, profile_value_item in list(fb.db.reference(f'account/{property_id}/profile/{each_pseudo}/{each_key}').get().items()): if profile_value_item['id'] == each_detail['id']: print(f"Deleted {each_detail['id']} in {each_key} from {each_pseudo}") if (fb.db.reference(f'account/{property_id}/profile/{each_pseudo}/{each_key}/{profile_key}').get()): fb.db.reference(f'account/{property_id}/profile/{each_pseudo}/{each_key}/{profile_key}').delete() # Unmerge all pseudo that was merged by this version_key for each_pseudo in list(log_details['merge'].keys()): for each_pseudo, details in log_details['merge'].items(): for key, detail in details.items(): print(f"Unmerging {detail['new']} from {detail['old']}") fb.db.reference(f"account/{property_id}/profile_temp/{detail['mapping']}/{detail['value']}").set(detail['old']) fb.db.reference(f"account/{property_id}/mapping/{detail['old']}").delete() ## mark version_key as deleted fb.db.reference(f'account/{property_id}/offline_log/{table_name}/{version_key}/status').set("deleted") bq = BigQuery() bq.delete_version(property_id, table_name, version_key) return {"status": f"Data in table {table_name} with version key '{version_key}' deleted successfully"}, 200 return {"status":"error",'message': "Method not allowed"}, 405 @schema_data.route('/get_data', methods=['GET']) def get_data(): if request.method == 'GET': param = request.args for req in ['property_id','table_name']: if req not in param: return {"status":"error",'message': f"{req} is require"}, 400 property_id = request.args.get('property_id') table_name = request.args.get('table_name') limit = request.args.get('limit', 50) bq = BigQuery() df = bq.get_data_with_limit(f"client_{property_id}", f"offline_{table_name}", limit) if df.empty: return {"status": "error", "message": "No data found"}, 404 ref = fb.db.reference(f'property/{property_id}/offline_mapping/offline_{table_name}/sensitive_data') sensitive_data = ref.get() if isinstance(sensitive_data, dict): sensitive_data_list = list(sensitive_data.values()) elif isinstance(sensitive_data, list): sensitive_data_list = sensitive_data else: sensitive_data_list = [] private_key = Key.load_key_from_env("private_key") if len(sensitive_data_list) > 0: df = Key.decrypt_dataframe_columns(df, sensitive_data_list, private_key) data = df.to_dict(orient='records') # Fix potential NumPy objects inside the records for row in data: for key, value in row.items(): if isinstance(value, np.ndarray): row[key] = value.tolist() elif isinstance(value, (np.integer, np.floating)): row[key] = value.item() return {"status": "ok", "message": "Data retrieved successfully", "data": data}, 200 return {"status":"error",'message': "Method not allowed"}, 405