from fastapi import Request, Query, Depends, Header, APIRouter, HTTPException from fastapi.responses import JSONResponse import pytz import os from datetime import datetime, timedelta from connectors.firebase.firebase import Firebase from api.feature.automation.node import * from api.feature.audience.audience_builder import AudinceBuilder 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_job_audience" def is_divisible_hour(n: int, hour_str: str) -> bool: # hour_str format: "HH:MM" hour = int(hour_str.split(":")[0]) return hour % n == 0 def handel_audience_recal(audience_context:dict): builder = AudinceBuilder(json=audience_context, cache=False) valid, message = builder._validate_json_data() if not valid: return {'status': 'error', 'message': message}, 500 final_result = builder.get_final_audience() total_audience = len(final_result) return final_result, total_audience def handel_set_new_audience_data(property_id, audience_id, final_result:list, total_audience:int): """ Update audience data after recalculation. For Preset audiences: Write to BigQuery audience_members table For Builder audiences: No changes (still calculates real-time) """ try: # Get audience context to determine type audience_context = fb.db.reference().child(f"account/{property_id}/audience/{audience_id}").get() if not audience_context: logging.error(f"Audience {audience_id} not found for property {property_id}") return False # Determine audience type audience_type = audience_context.get('type') if not audience_type: # Fallback: determine type from structure if 'nodes' in audience_context and 'connections' in audience_context: audience_type = 'builder' elif 'audience_syntax' in audience_context or 'audience_name' in audience_context: audience_type = 'preset' else: audience_type = 'builder' # Default to builder if unclear now = datetime.now(timezone_utc) # For Preset: Write to BigQuery audience_members table if audience_type == 'preset': from connectors.bigquery.bq import BigQuery bq = BigQuery() dataset_id = f"client_{property_id}" project_id = os.environ.get('GCP_PROJECT', 'customer-360-profile') table_ref = f"{project_id}.{dataset_id}.audience_members" try: # Mark old members as inactive safe_audience_id = audience_id.replace("'", "''") update_query = f""" UPDATE `{table_ref}` SET is_active = FALSE, removed_at = CURRENT_DATETIME() WHERE audience_id = '{safe_audience_id}' AND is_active = TRUE """ # Execute UPDATE query (doesn't return rows) query_job = bq.client.query(update_query) query_job.result() # Wait for completion # Insert new members if final_result and len(final_result) > 0: rows_to_insert = [] for user_pseudo_id in final_result: if user_pseudo_id: # Skip None or empty values rows_to_insert.append({ 'audience_id': audience_id, 'user_pseudo_id': str(user_pseudo_id), 'added_at': now.isoformat(), 'removed_at': None, 'is_active': True }) if rows_to_insert: errors = bq.load_data(target_table=table_ref, data=rows_to_insert) if errors: logging.error(f"BigQuery insert errors for audience {audience_id}: {errors}") return False logging.info(f"Updated BigQuery for Preset audience {audience_id}: {len(final_result)} members") except Exception as e: logging.error(f"Error updating BigQuery for Preset audience {audience_id}: {e}") # Fallback: write to Firebase if BigQuery fails fb.db.reference().child(f"account/{property_id}/audience/{audience_id}/user_pseudo_ids").set(final_result) logging.warning(f"Fell back to Firebase for audience {audience_id}") # For Builder: No changes needed (still calculates real-time) # Do NOT write user_pseudo_ids array to Firebase (no array to store) # Update Firebase metadata only (for both Preset and Builder) fb.db.reference().child(f"account/{property_id}/audience/{audience_id}/total_audience").set(int(total_audience)) fb.db.reference().child(f"account/{property_id}/audience/{audience_id}/lastupdate").set(now.isoformat()) return True except Exception as e: logging.error(f"Error in handel_set_new_audience_data for audience {audience_id}: {e}") return False @router.post("/calculate", description="Scheduled job to calculate dynamic audiences") async def post_api_job_audience_calculate(default_request: Request): try: date_str = datetime.now(timezone_bkk).strftime("%Y%m%d%H00") now_hour = datetime.now(timezone_bkk).strftime("%H:00") prefix_logging = "audience_scheduled_calculate" logging.info(f"{prefix_logging}: START JOB") fb_ref = fb.db.reference() accounts = fb_ref.child("account").get(shallow=True) for account in accounts: log_job = [] audiences = fb_ref.child(f"account/{account}/audience/").get(shallow=True) if not audiences: continue for audience in audiences: is_active = fb_ref.child(f"account/{account}/audience/{audience}/active").get() if not is_active: continue setting = fb_ref.child(f"account/{account}/audience/{audience}/settings").get() cal_type = setting['type'] if cal_type != 'dynamic': continue calculation = setting['calculation'] calculation_type = calculation['type'] calculation_time = calculation['time'] if calculation_type == 'daily': if now_hour == calculation_time: #Re-cal audience_context = fb_ref.child(f"account/{account}/audience/{audience}").get() final_result, total_audience = handel_audience_recal(audience_context) result = handel_set_new_audience_data(account, audience, final_result, total_audience) log_job.append( { "property_id": account, "audience_id": audience, "status": result, } ) fb_ref.child(f"account/{account}/audience/{audience}/lastupdate").set(datetime.now(timezone_utc).isoformat()) elif calculation_type == 'hourly': if is_divisible_hour(int(calculation_time)): #Re-cal audience_context = fb_ref.child(f"account/{account}/audience/{audience}").get() final_result, total_audience = handel_audience_recal(audience_context) result = handel_set_new_audience_data(account, audience, final_result, total_audience) log_job.append( { "property_id": account, "audience_id": audience, "status": result, } ) fb_ref.child(f"account/{account}/audience/{audience}/lastupdate").set(datetime.now(timezone_utc).isoformat()) fb_ref.child(f"account/{account}/job/log/audience/{date_str}").set(log_job) logging.info(f"{prefix_logging}: END JOB") return JSONResponse(status_code=200, content={"status": "ok", "message": "Audience calculation job completed"}) except Exception as e: logging.error(f"Error parsing {LOGGING_PREFIX}: {e}") return JSONResponse(status_code=500, content={'status': 'error', 'message': str(e)})