from fastapi import Request, Query, Depends, Header, APIRouter, HTTPException from fastapi.responses import JSONResponse import pytz import os import uuid from datetime import datetime, timedelta from connectors.firebase.firebase import Firebase import utility.function as func from connectors.bigquery.bq import BigQuery from utility.offline import Offline from utility.function import Function, Pack, Key, Profile from models.offline_data import * import logging logging.basicConfig(level=logging.INFO) timezone_utc = pytz.utc timezone_bkk = pytz.timezone('Asia/Bangkok') fb = Firebase(host=os.environ.get("FIREBASE_HOST")) router = APIRouter() LOGGING_PREFIX = "api_offline_data" @router.post("/insert_data",description="Insert offline data into customer profiles") async def insert_offline_data(request: InsertOfflineDataRequest, default_request: Request): import pandas as pd try: df = pd.DataFrame(request.data) property_id = request.property_id table_name = request.table_name demographic = fb.db.reference(f'property/{request.property_id}/offline_mapping/offline_{request.table_name}/demographic_col').get() contact_key = fb.db.reference(f'property/{request.property_id}/offline_mapping/offline_{request.table_name}/contact_key').get() other_key = fb.db.reference(f'property/{request.property_id}/offline_mapping/offline_{request.table_name}/other_key').get() event_mapping = fb.db.reference(f'property/{request.property_id}/offline_mapping/offline_{request.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/{request.property_id}/offline_mapping/offline_{request.table_name}/sensitive_data').get() key_columns = list(fb.db.reference(f'property/{request.property_id}/offline_mapping/offline_{request.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/{request.property_id}/offline_mapping/offline_{request.table_name}/contact_key/Phone').get() email_col = fb.db.reference(f'property/{request.property_id}/offline_mapping/offline_{request.table_name}/contact_key/Email').get() customer_id_col = fb.db.reference(f'property/{request.property_id}/offline_mapping/offline_{request.table_name}/other_key/CustomerID').get() # Has Phone column if phone_col: profile_temp_phone = fb.db.reference(f'account/{request.property_id}/profile_temp/phoneNumber').get() or {} profile_temp_email = fb.db.reference(f'account/{request.property_id}/profile_temp/email').get() or {} profile_temp_customer_id = fb.db.reference(f'account/{request.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/{request.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/{request.property_id}/profile_temp/email/{hash_email}').set(profile_temp_phone[hash_phone]) fb.db.reference(f'account/{request.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/{request.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/{request.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/{request.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/{request.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/{request.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/{request.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/{request.property_id}/profile_temp/phoneNumber/{hash_phone}').set(new_pseudo) fb.db.reference(f'account/{request.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/{request.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/{request.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/{request.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/{request.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/{request.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/{request.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/{request.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/{request.property_id}/profile_temp/email').get() or {} profile_temp_customer_id = fb.db.reference(f'account/{request.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/{request.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/{request.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/{request.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/{request.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/{request.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/{request.property_id}/offline_mapping/offline_{request.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_{request.property_id}",f"offline_{request.table_name}_temp") bq.load_data_df(f"client_{request.property_id}",f"offline_{request.table_name}_temp",df_encrypted) bq.delete_when_match("customer-360-profile",f"client_{request.property_id}",f"offline_{request.table_name}", f"client_{request.property_id}",f"offline_{request.table_name}_temp", f"ON {on_clause} ") bq.load_data_df(f'client_{request.property_id}',f"offline_{request.table_name}",df_encrypted) count_row = bq.bq_count_rows(f"offline_{request.table_name}") fb.db.reference(f'property/{request.property_id}/offline_mapping/offline_{request.table_name}/row_count').set(count_row) return JSONResponse(status_code=200, content={"status": "Data inserted successfully"}) except Exception as e: logging.error(f"Error parsing {LOGGING_PREFIX}_{default_request.path_params}: {e}") return JSONResponse(status_code=500, content={'status': 'error', 'message': str(e)}) @router.post("/truncate_table",description="Truncate offline table data") async def truncate_offline_table(request: OfflineTableBase, default_request: Request): try: bq = BigQuery() bq.delete_data("customer-360-profile",f"client_{request.property_id}",f"offline_{request.table_name}") return JSONResponse(status_code=200, content={"status": f"Offline table {request.table_name} truncated successfully"}) except Exception as e: logging.error(f"Error parsing {LOGGING_PREFIX}_{default_request.path_params}: {e}") return JSONResponse(status_code=500, content={'status': 'error', 'message': str(e)}) @router.post("/delete_data",description="Delete offline data by version key") async def delete_offline_data(request: DeleteOfflineDataRequest, default_request: Request): try: property_id = request.property_id table_name = request.table_name version_key = request.version_key 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 JSONResponse(status_code=200, content={"status": f"Data in table {table_name} with version key '{version_key}' deleted successfully"}) except Exception as e: logging.error(f"Error parsing {LOGGING_PREFIX}_{default_request.path_params}: {e}") return JSONResponse(status_code=500, content={'status': 'error', 'message': str(e)}) @router.get("/get_data", response_model=DataResponse, description="Get offline data from table") async def get_offline_data( property_id: str = Query(..., description="Property ID"), table_name: str = Query(..., description="Offline Table Name"), limit: int = Query(50, description="Number of records to retrieve"), default_request: Request = None ): try: import numpy as np bq = BigQuery() df = bq.get_data_with_limit(f"client_{property_id}", f"offline_{table_name}", limit) if df.empty: return JSONResponse(status_code=200, content={"data": []}) 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 JSONResponse(status_code=200, content={"status": "ok", "message": "Data retrieved successfully", "data": data}) except Exception as e: logging.error(f"Error parsing {LOGGING_PREFIX}_{default_request.path_params}: {e}") return JSONResponse(status_code=500, content={'status': 'error', 'message': str(e)})