import os import re import pytz import requests import pandas as pd from connectors.firebase.firebase import Firebase from datetime import datetime, timedelta from utility.function import Key, Function fb = Firebase(host=os.environ.get("FIREBASE_HOST")) class Offline: def check_existing_pseudo_detail(property_id, psudeo_id, key, value): existing = fb.db.reference(f'account/{property_id}/profile/{psudeo_id}/{key}').get() if existing: for field, col in (existing or {}).items(): if col['id'] == value: print(value, " = ", col['id']) return True else: return False else: return False def log_merge(property_id, filename, version_key, old_id, new_id, contact_type, value): fb.db.reference( f"account/{property_id}/offline_log/{filename}/{version_key}/merge/{new_id}" ).push({ "old": old_id, "new": new_id, "mapping": contact_type, "value": value, "created_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "updated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S") }) def log_change_customer_pseudo(property_id, filename, version_key, old_id, new_id, contact_type, value): fb.db.reference( f"account/{property_id}/offline_log/{filename}/{version_key}/CustomerID/{new_id}" ).push({ "old": old_id, "new": new_id, "mapping": contact_type, "value": value, "created_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "updated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S") }) ## add_details_and_log(property_id, table_name, version_key, phone_psuedo, hash_email, "email") def add_details_and_log(table_name, property_id, filename, version_key, pseudo_id, info, key): status = Offline.check_existing_pseudo_detail(property_id, pseudo_id, key, info) if status == False: temp_detail = { "id": info, "source": { "from": table_name, "id": info, "page_id": version_key }, "created_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "updated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S") } print(f"Adding {key} to pseudo_id {pseudo_id}") fb.db.reference(f"account/{property_id}/profile/{pseudo_id}/{key}").push(temp_detail) fb.db.reference(f"account/{property_id}/offline_log/{filename}/{version_key}/add_details/{pseudo_id}/{key}").push(temp_detail) else: print(f"Pseudo ID {pseudo_id} already has {key} : {info}, skipping log_add_details.") def log_gen_pseudo(property_id, table_name, version_key, contact_key, hash_contact, pseudo_id): fb.db.reference( f"account/{property_id}/offline_log/{table_name}/{version_key}/gen_pseudo/{pseudo_id}/{contact_key}" ).set({ "id": hash_contact, "source": { "from": table_name, "id": hash_contact, "page_id": version_key }, "created_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "updated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S") }) fb.db.reference(f"account/{property_id}/profile/{pseudo_id}").set({ "created_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "updated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "user_pseudo_id": pseudo_id }) fb.db.reference(f"account/{property_id}/profile/{pseudo_id}/{contact_key}").push({ "created_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "updated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "id": hash_contact, "source" : { "from": table_name, "id": hash_contact, "page_id": version_key } }) return True def log_mapped_fields(property_id, filename, version_key, pseudo_id, row, mapping_dict,remaining_sensitive=None,public_key=None,store_plain_for_sensitive=False): remaining_sensitive = set(remaining_sensitive or []) for field, col in (mapping_dict or {}).items(): if col in row: val = row[col] if pd.notna(val) and str(val).strip() and str(val).lower() != "nan": val_str = str(val) if col in remaining_sensitive: # Encrypt sensitive value if public_key is None: raise ValueError("public_key is required to encrypt sensitive fields") # Optionally store plaintext first (usually False for sensitive) if store_plain_for_sensitive: Offline.add_details_and_log(property_id, filename, version_key, pseudo_id, val_str, field) # Always store encrypted under _PGP enc = Key.pgp_encrypt(val_str, public_key) Offline.add_details_and_log(property_id, filename, version_key, pseudo_id, enc, f"{field}_PGP") else: # Non-sensitive: store as-is Offline.add_details_and_log(property_id, filename, version_key, pseudo_id, val_str, field) def delete_log_offline(property_id, filename, version_key): fb.db.reference( f"account/{property_id}/offline_log/{filename}/{version_key}" ).delete() def is_sha256_hex(s: str) -> bool: SHA256_HEX = re.compile(r"^[a-f0-9]{64}$") return bool(SHA256_HEX.fullmatch(str(s).strip().lower())) def normalize_phone(phone: str) -> str: """Return SHA-256 hex of phone unless it's already a 64-char hex hash.""" phone = str(phone).strip() return phone if Offline.is_sha256_hex(phone) else Function.hash256(phone) def normalize_email(email: str) -> str: """Return SHA-256 hex of email (lowercased) unless it's already a 64-char hex hash.""" email = str(email).strip() if Offline.is_sha256_hex(email): return email # common normalization before hashing return Function.hash256(email.lower()) def _as_ids(v): """Return a list of ids from any shape (scalar / dict / list / dict-of-children).""" if v is None: return [] if isinstance(v, (str, int, float, bool)): return [str(v)] if isinstance(v, list): out = [] for x in v: out.append(str(x.get("id")) if isinstance(x, dict) and "id" in x else str(x)) return [s for s in map(str, out) if s] if isinstance(v, dict): if "id" in v: return [str(v["id"])] # dict-of-children out = [] for child in v.values(): if isinstance(child, dict) and "id" in child: out.append(str(child["id"])) else: out.append(str(child)) return [s for s in map(str, out) if s] return [str(v)] def copy_profile_with_logging(property_id, filename, version_key, old_pseudo, new_pseudo, keys=None): """ Copy (NOT move) details from old_pseudo -> new_pseudo using your add_details_and_log(). - Skips duplicates automatically via check_existing_pseudo_detail inside add_details_and_log - Leaves old profile untouched """ if old_pseudo == new_pseudo: return src = fb.db.reference(f"account/{property_id}/profile/{old_pseudo}").get() or {} if not isinstance(src, dict): return # Choose which keys to copy if keys is None: keys = [k for k in src.keys() if k != "user_pseudo_id"] dst_base = fb.db.reference(f"account/{property_id}/profile/{new_pseudo}") for key in keys: val = src.get(key) if val is None: continue # For timestamps: only set if destination missing (no logging needed) if key in ("created_at", "createdAt", "updated_at", "updatedAt"): if dst_base.child(key).get() is None: dst_base.child(key).set(str(val)) continue # For detail collections: use your existing add_details_and_log (does dedupe + logs) for _id in Offline._as_ids(val): if _id: Offline.add_details_and_log(property_id, filename, version_key, new_pseudo, _id, key) def build_on(cols, left='ori', right='temp', null_safe=True): op = '=' if null_safe else '=' # null-safe in BigQuery return ' AND '.join([f'{left}.`{c}` {op} {right}.`{c}`' for c in cols])