from fastapi import Request, Query, Depends, Header, APIRouter, HTTPException from fastapi.responses import JSONResponse from connectors.firebase.firebase import Firebase import pytz import os import json from datetime import datetime, timedelta from models.dashboard import * from utility.dashboard.search import C360Search from api.dashboard.profile import fucntion as profileFunction import logging logging.basicConfig(level=logging.INFO) timezone_utc = pytz.utc fb = Firebase(host=os.environ.get("FIREBASE_HOST")) router = APIRouter() LOGGING_PREFIX = "api_dashboard_profile" @router.get("/explore", response_model=GetProfileExploreRespone, description="Get profile explore data") async def get_dashboard_profile_explore( default_request: Request, property_id: str = Query(..., description="Property ID"), key: str = Query(..., description="Search key"), value: str = Query(..., description="Search value"), ): try: if key != 'user_pseudo_id': c360search = C360Search(property_id=property_id) user_pseudo_id = c360search.get_profile(key, value) else: user_pseudo_id = value if not user_pseudo_id: return JSONResponse(status_code=404,content={"status": "not_found", "message": 'User not fond in this property'}) import concurrent.futures from connectors.looker.lookerSDK import LookerSDK import pandas as pd looker = LookerSDK(property_id=property_id) #start query full_result = [] queries = { "lifetimeSpend": { "view": "profile_explore", "fields": ["transaction.total_price"], "filters": {"profile_explore.user_pseudo_id": user_pseudo_id}, "model": f"c360_client_{property_id}", "limit": "1" }, "name": { "view": "profile_explore", "fields": ["profile_explore__user_property.value_string"], "filters": { "profile_explore__user_property.key": "FirstName", "profile_explore.user_pseudo_id": user_pseudo_id }, "model": f"c360_client_{property_id}", "limit": "1" }, "transaction": { "view": "profile_explore", "fields": [ "transaction.receipt_date_date", "transaction.receipt_no", "transaction.item_name", "transaction.total_price" ], "filters": { "transaction.receipt_no": "-NULL", "profile_explore.user_pseudo_id": user_pseudo_id }, "sorts": ["transaction.receipt_date_date"], "model": f"c360_client_{property_id}", "limit": "-1" }, "transaction_overview": { "view": "profile_explore", "fields": [ "transaction.total_price", "transaction.total_order", "transaction.average_order_amount", ], "filters": { "transaction.receipt_no": "-NULL", "profile_explore.user_pseudo_id": user_pseudo_id }, "model": f"c360_client_{property_id}", "limit": "1" }, "keywords": { "view": "profile_explore", "fields": [ "profile_explore.keywords_string" ], "filters": { "profile_explore.user_pseudo_id": user_pseudo_id }, "model": f"c360_client_{property_id}", "limit": "1" }, "branch": { "view": "event_offline", "fields": [ "event_offline__event_property.value" ], "filters": { "event_offline__event_property.key": "%branch%", "event_offline.user_pseudo_id": user_pseudo_id }, "filter_expression": "matches_filter(${event_offline__event_property.key}, `-%code%`)", "model": f"c360_client_{property_id}", "limit": "1" } } def run_looker_query(query_obj): try: query = looker.sdk.create_query(query_obj) result = looker.sdk.run_query(query.id, result_format="json", cache=True) return json.loads(result) except: return [] def get_user_profile(): user_profile = fb.db.reference().child(f"account/{property_id}/profile/{user_pseudo_id}").get() return profileFunction.transformUserData_main_profile(user_profile) def get_journey(): return profileFunction.transformUserData_journey(property_id=property_id, user_pseudo_id=user_pseudo_id) def get_second_profile(): return profileFunction.transformUserData_second_profile() def get_channel_engagement(): return profileFunction.transformUserData_channel_engagement(property_id=property_id, user_pseudo_id=user_pseudo_id) def get_conversation(): return profileFunction.transformUserData_conversation(property_id=property_id, user_pseudo_id=user_pseudo_id) with concurrent.futures.ThreadPoolExecutor(max_workers=6) as executor: futures = { "user_profile": executor.submit(get_user_profile), "journey": executor.submit(get_journey), "second_profile": executor.submit(get_second_profile), "channel_engagement": executor.submit(get_channel_engagement), "conversation": executor.submit(get_conversation), "lifetimeSpend": executor.submit(run_looker_query, queries["lifetimeSpend"]), "name": executor.submit(run_looker_query, queries["name"]), "transaction": executor.submit(run_looker_query, queries["transaction"]), "transaction_overview": executor.submit(run_looker_query, queries["transaction_overview"]), "keywords": executor.submit(run_looker_query, queries["keywords"]), "branch": executor.submit(run_looker_query, queries["branch"]), } results = {k: f.result() for k, f in futures.items()} # --- Process results --- user_profile_object = results["user_profile"] # Enrich with lifetime spend lifetimeSpend = results["lifetimeSpend"] if lifetimeSpend and isinstance(lifetimeSpend, list): user_profile_object['data']['calculate']['lifetime_spend'] = lifetimeSpend[0].get('transaction.total_price', 0) # Enrich with name name = results["name"] if name and isinstance(name, list): user_profile_object['data']['name'] = name[0].get('profile_explore__user_property.value_string', '-') #Enrich Keyword keywords = results['keywords'] if keywords and isinstance(keywords, list): results['second_profile']['data']['interests'] = keywords #Enrich branch branch = results['branch'] if branch and isinstance(branch, list): results['second_profile']['data']['branch'] = branch # Transaction data transaction_object = { "id": "transaction", "life_time_order": results['transaction_overview'][0]['transaction.total_order'] if len(results['transaction_overview'])>0 else '-', "lifetime_spend": results['transaction_overview'][0]['transaction.total_price'] if len(results['transaction_overview'])>0 else '-', "average_order_amount": results['transaction_overview'][0]['transaction.average_order_amount'] if len(results['transaction_overview'])>0 else '-', "data": results["transaction"] } # Combine results full_result.extend([ user_profile_object, results["journey"], results["second_profile"], results["channel_engagement"], results["conversation"], transaction_object ]) final_object = { "status": "ok", "user_pseudo_id": user_pseudo_id, "widgets": full_result } logging.info("dashboard_profile_explore complete") 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("/detail/journey", description="Get detailed journey profile data") async def get_dashboard_profile_journey_detail( default_request: Request, property_id: str = Query(..., description="Property ID"), user_pseudo_id: str = Query(..., description="User Pseudo ID") ): try: object = profileFunction.transformUserData_detail_journey(property_id, user_pseudo_id) if 'data' not in object: object['status'] = 'not_fond' return JSONResponse(status_code=404,content=object) else: object['status'] = 'ok' return object 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("/detail/profile", description="Get detailed profile data") async def get_dashboard_profile_profile_detail( default_request: Request, property_id: str = Query(..., description="Property ID"), user_pseudo_id: str = Query(..., description="User Pseudo ID") ): try: returnObject = profileFunction.transformUserData_detail_profile(property_id, user_pseudo_id, fb) returnObject['status'] = 'ok' return returnObject 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("/list", response_model=GetListProfileRespone, description="Get list of profiles") async def get_dashboard_profile_list( default_request: Request, property_id: str = Query(..., description="Property ID"), query: str = Query(..., description="Search query"), start: str = Query(..., description="Date in format YYYY/MM/DD"), end: str = Query(..., description="Date in format YYYY/MM/DD"), page: int = Query(1, description="Page number"), page_size: int = Query(50, description="Number of items per page") ): try: if start != None and '/' not in start: return JSONResponse(status_code=400, content={'message': f"parameters start must be in format YYYY/MM/DD"}) if end != None and '/' not in end: return JSONResponse(status_code=400, content={'message': f"parameters end must be in format YYYY/MM/DD"}) dateNow = datetime.now().strftime('%Y/%m/%d') dateNow_1 = (datetime.now() - timedelta(days=1)).strftime('%Y/%m/%d') from connectors.looker.lookerSDK import LookerSDK import pandas as pd looker = LookerSDK(property_id=property_id) #Handle date if start and end: if start == end: datefilter = start else: datefilter = f"{start} to {end}" else: datefilter = "1 year" query = { "view": "profile_explore", "fields": [ "profile_explore.user_pseudo_id_list" ], "filter_expression": f'''(matches_filter(${{profile_explore__user_property.value_string}}, `%{query.replace("_", "^_")}%`) AND matches_filter(${{profile_explore.lastupdate_date}}, `{datefilter}`)) OR (matches_filter(${{profile_explore.keywords_string}}, `%{params.query.replace("_", "^_")}%`) AND matches_filter(${{profile_explore.lastupdate_date}}, `{datefilter}`)) OR (matches_filter(${{profile_explore.user_pseudo_id}}, `%{params.query.replace("_", "^_")}%`) AND matches_filter(${{profile_explore.lastupdate_date}}, `{datefilter}`))''', "model": f"c360_client_{property_id}", "limit": "-1" } data = looker.sdk.run_query(looker.sdk.create_query(query).id, result_format="json") data = json.loads(data) if not data: return JSONResponse(status_code=404, content={'status': 'not_fond', 'message': f"{query} not match any result"}) userList = data[0]['profile_explore.user_pseudo_id_list'] if not userList: return JSONResponse(status_code=404, content={'status': 'not_fond', 'message': f"{query} not match any result"}) userListString = ",".join(userList) queryProfile = { "view": "profile_explore", "fields": [ "profile_explore.user_pseudo_id", "profile_explore.lastupdate_time", "profile_explore.user_property", "profile_explore.keywords_string" ], "filters": { "profile_explore.user_pseudo_id": userListString }, "sorts": [ "profile_explore.lastupdate_time desc" ], "model": f"c360_client_{property_id}", "limit": "-1" } dataProfile = looker.sdk.run_query(looker.sdk.create_query(queryProfile).id, result_format="json") dataProfile = json.loads(dataProfile) df =pd.DataFrame(dataProfile) df['profile_explore.user_property'] = df['profile_explore.user_property'].apply(json.loads) df.rename(columns={ "profile_explore.user_pseudo_id": "user_pseudo_id", "profile_explore.lastupdate_time": "lastupdate", "profile_explore.user_property": "user_property", "profile_explore.keywords_string": "keywords", }) start_page = (page - 1) * page_size end_page = start_page + page_size df_page = df.iloc[start_page:end_page] total_items = len(df) total_pages = (total_items + page_size - 1) // page_size records = df_page.to_dict(orient='records') return {'status': 'ok', 'data': records, 'page': page, 'page_size': page_size, 'total_pages': total_pages, 'total_records': total_items} 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("/search", response_model=GetListProfileRespone, description="Search profiles") async def get_dashboard_profile_search( default_request: Request, property_id: str = Query(..., description="Property ID"), query: str = Query(..., description="Search query"), start: str = Query(..., description="Date in format YYYY/MM/DD"), end: str = Query(..., description="Date in format YYYY/MM/DD"), page: int = Query(1, description="Page number"), page_size: int = Query(50, description="Number of items per page") ): try: from connectors.looker.lookerSDK import LookerSDK import pandas as pd looker = LookerSDK(property_id=property_id) #Handle date if start and end: if start == end: datefilter = start else: datefilter = f"{start} to {end}" else: datefilter = "1 year" query = { "view": "profile_explore", "fields": [ "profile_explore.user_pseudo_id_list" ], "filter_expression": f'''(matches_filter(${{profile_explore__user_property.value_string}}, `%{query.replace("_", "^_")}%`) AND matches_filter(${{profile_explore.lastupdate_date}}, `{datefilter}`)) OR (matches_filter(${{profile_explore.keywords_string}}, `%{query.replace("_", "^_")}%`) AND matches_filter(${{profile_explore.lastupdate_date}}, `{datefilter}`)) OR (matches_filter(${{profile_explore.user_pseudo_id}}, `%{query.replace("_", "^_")}%`) AND matches_filter(${{profile_explore.lastupdate_date}}, `{datefilter}`))''', "model": f"c360_client_{property_id}", "limit": "-1" } data = looker.sdk.run_query(looker.sdk.create_query(query).id, result_format="json") data = json.loads(data) if not data: return JSONResponse(status_code=404, content={'status': 'not_fond', 'message': f"{query} not match any result"}) userList = data[0]['profile_explore.user_pseudo_id_list'] if not userList: return JSONResponse(status_code=404, content={'status': 'not_fond', 'message': f"{query} not match any result"}) userListString = ",".join(userList) queryProfile = { "view": "profile_explore", "fields": [ "profile_explore.user_pseudo_id", "profile_explore.lastupdate_time", "profile_explore.user_property", "profile_explore.keywords_string" ], "filters": { "profile_explore.user_pseudo_id": userListString }, "sorts": [ "profile_explore.lastupdate_time desc" ], "model": f"c360_client_{property_id}", "limit": "-1" } dataProfile = looker.sdk.run_query(looker.sdk.create_query(queryProfile).id, result_format="json") dataProfile = json.loads(dataProfile) df =pd.DataFrame(dataProfile) df['profile_explore.user_property'] = df['profile_explore.user_property'].apply(json.loads) df.rename(columns={ "profile_explore.user_pseudo_id": "user_pseudo_id", "profile_explore.lastupdate_time": "lastupdate", "profile_explore.user_property": "user_property", "profile_explore.keywords_string": "keywords", }) start_page = (page - 1) * page_size end_page = start_page + page_size df_page = df.iloc[start_page:end_page] total_items = len(df) total_pages = (total_items + page_size - 1) // page_size records = df_page.to_dict(orient='records') return {'status': 'ok', 'data': records, 'page': page, 'page_size': page_size, 'total_pages': total_pages, 'total_records': total_items} 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)})