from fastapi import Request, Query, Depends, Header, APIRouter, HTTPException, UploadFile, File, Form from fastapi.responses import JSONResponse from utility.function import Function from connectors.firebase.firebase import Firebase from utility.authorization import TokenAuthorization from connectors.cloudStorage.bucketstorage import Storage import pytz import os import uuid from datetime import datetime from werkzeug.utils import secure_filename from models.property import * import logging logging.basicConfig(level=logging.INFO) timezone_utc = pytz.utc fb = Firebase(host=os.environ.get("FIREBASE_HOST")) router = APIRouter() LOGGING_PREFIX = "api_property" @router.post("", response_model=PropertyRegisterRespone, description="Register a new property") async def property_register_post(request: PropertyRegister, default_request: Request = None): user_id = Function.getUserIDFromHeader(default_request.headers) try: now = datetime.now(timezone_utc) time = now.strftime('%Y-%m-%d %H:%M:%S') request.createdate, request.lastupdate = time, time id = Function.text_to_numeric_token(request.name) #Check exsing propertyEx = fb.db.reference().child(f"property/{id}").get(shallow=True) if propertyEx: return JSONResponse(status_code=400,content={"status": "error", "message": 'This Property already created'}) fb.db.reference().child(f"property/{id}").set(request.model_dump()) #Set user data fb.db.reference().child(f"""users/{user_id}/property/{id}/role""").set("owner") #Add channel to master masterUser = fb.get_by_child(child=f"users/master") if masterUser: listUserMaster = list(masterUser) for user in listUserMaster: try: jobMaster = fb.db.reference().child(f"""users/{user}/property/{id}/role""").update("master") except: jobMaster = fb.db.reference().child(f"""users/{user}/property/{id}/role""").set("master") # Task Create dataset from connectors.bigquery.bq import BigQuery, Table bq = BigQuery() dSet = bq.create_dataset(f"client_{id}") if dSet: #Create event table bq.create_table_from_json_schema(f"client_{id}", "event", Table.tableEvent(), "eventTimeStamp", cluster_fields=["pageId","eventName", "user_pseudo_id", "source"]) bq.create_table_from_json_schema(f"client_{id}", "customer_profile", Table.tableCustomerProfile(), "lastupdate", cluster_fields=["user_pseudo_id"]) bq.create_table_from_json_schema(f"client_{id}", "customer_profile_temp", Table.tableCustomerProfile(), "lastupdate", cluster_fields=["user_pseudo_id"]) bq.create_table_from_json_schema(f"client_{id}", "profile_mapping", Table.tableProfileMapping(), "lastupdate", cluster_fields=["user_pseudo_id","unified_user_pseudo_id"]) bq.create_table_from_json_schema(f"client_{id}", "profile_mapping_temp", Table.tableProfileMapping(), "lastupdate", cluster_fields=["user_pseudo_id","unified_user_pseudo_id"]) bq.create_table_from_json_schema(f"client_{id}", "user", Table.tebleUserProfile(), "lastupdate", cluster_fields=["user_pseudo_id"]) bq.create_table_from_json_schema(f"client_{id}", "user_temp", Table.tebleUserProfile(), "lastupdate", cluster_fields=["user_pseudo_id"]) bq.create_table_from_json_schema(f"client_{id}", "profile_explore", Table.tableProfileExplore(), "lastupdate", cluster_fields=["user_pseudo_id"]) bq.create_table_from_json_schema(f"client_{id}", "profile_explore_temp", Table.tableProfileExplore(), "lastupdate", cluster_fields=["user_pseudo_id"]) return {"status":"ok",'message': 'Success', 'property_id': str(id)} 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("", response_model=PropertyRegisterRespone, description="Get property information") async def property_register_get( default_request: Request = None, property_id: str = Query(..., description="Property ID or 'all' to get all properties") ): try: local_key = default_request.headers.get('Authorization') bearer = local_key.replace("Bearer ", "") userAuth = TokenAuthorization(access_token=bearer) userData = userAuth.validate() #Check user property try: userPropertyList = userData['property'] except: return {'status': 'error', 'message': "User don't have any property"},400 if property_id != "all" and property_id in userPropertyList: property_data = fb.get_by_child(child=f"property/{property_id}") logging.info(f"Property data: {property_data}") if property_data: return {"status":"ok",'message': 'Success', 'data': [property_data]} else: return {"status":"error",'message': 'Not found'}, 404 elif property_id == 'all': list_data = [] for prop in userPropertyList: property_data = fb.get_by_child(child=f"property/{prop}") list_data.append(property_data) if list_data: return {"status":"ok",'message': 'Success', 'data': list_data}, 200 return JSONResponse(status_code=404,content={"status": "not_found", "message": 'You not have permission to this property'}) 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("/image/upload", response_model=PropertyImageUploadRespone, description="Upload property image") async def property_image_upload( default_request: Request, file: UploadFile = File(...), property_id: str = Form(...) ): try: if file.content_type != "image/jpeg": raise HTTPException(400, detail="Content-Type must be image/jpeg") image_data = await file.read() if not image_data: raise HTTPException(400, detail="No selected image") now = datetime.now(timezone_utc) time_now = int(now.timestamp()) generate_image_file_name = f"{uuid.uuid4()}_{time_now}.png" filename = secure_filename(generate_image_file_name) gcs_blob_name = f"property/{filename}" try: storage = Storage(project_id=os.environ.get("GCP_PROJECT"),bucket_name=os.environ.get("GC_BUCKET_NAME")) link = storage.upload_from_byte(file_path=gcs_blob_name, image_bytes=image_data) if not link: raise HTTPException(500, detail="Failed to upload image to GCS") fb.db.reference().child(f"property/{property_id}/image").set(link) return {"status": "ok", "message": "Image uploaded successfully", "image_address": link} except Exception as e: logging.exception("Error processing upload") raise HTTPException(500, detail=str(e)) except HTTPException: raise except Exception as e: logging.error(f"Error parsing {LOGGING_PREFIX}_{default_request.path_params}: {e}") raise HTTPException(500, detail=str(e)) @router.get("/list-channel", response_model=PropertyListChannelRespone, description="List channels of a property") async def property_list_channels( default_request: Request, property_id: str = Query(..., description="Property ID"), channel_type: str = Query(..., description="Channel type or 'all' to get all channels") ): try: if channel_type == 'all': channels = fb.db.reference().child(f"property/{property_id}/channel").get() if not channels: return JSONResponse(status_code=404,content={"status": "error", "message": 'This Property has no channel'}) result = [] for channel, accounts in channels.items(): channel_data = [] for account_id, info in accounts.items(): entry = {"id": account_id} if "name" in info: entry["name"] = info["name"] channel_data.append(entry) result.append({"channel": channel, "data": channel_data}) return {'status': 'ok', 'data': result} else: channels = fb.db.reference().child(f"property/{property_id}/channel/{channel_type}").get() if not channels: return {'status': 'error', 'message': f'This Property has no {channel_type} subscribe'}, 404 channel_data = [] for account_id, info in channels.items(): entry = {"id": account_id} if "name" in info: entry["name"] = info["name"] channel_data.append(entry) return {'status': 'ok', 'data': channel_data} except Exception as e: logging.error(f"Error parsing {LOGGING_PREFIX}_{default_request.path_params}: {e}") raise HTTPException(500, detail=str(e)) @router.post("/admin/phone", description="Add, replace, or remove admin phone numbers for a property") async def property_add_admin_phone(default_request: Request, request: PropertyAdminPhone): from utility.adminPhoneService import (add_admin_phones, replace_admin_phones, remove_admin_phones) try: try: if request.action == "add": env_key, out = add_admin_phones(request.property_id, request.phones) elif request.action in ("replace", "edit"): env_key, out = replace_admin_phones(request.property_id, request.phones) elif request.action == "remove": env_key, out = remove_admin_phones(request.property_id, request.phones) else: return {"status": "error","message": "action must be one of: add, replace, remove"} return { "status": "success", "property_id": request.property_id, "env_key": env_key, "action": request.action, "phones": out } except ValueError as ve: return JSONResponse(status_code=400,content={"status": "error", "message": str(ve)}) except Exception as e: return JSONResponse(status_code=500,content={"status": "error", "message": str(ve)}) except Exception as e: logging.error(f"Error parsing {LOGGING_PREFIX}_{default_request.path_params}: {e}") raise HTTPException(500, detail=str(e)) @router.get("/admin/phone", description="Get admin phone numbers for a property") async def property_get_admin_phone( default_request: Request, property_id: str = Query(..., description="Property ID"), ): from utility.adminPhoneService import (get_admin_phones) try: try: env_key, phones, _ = get_admin_phones(property_id) return {"status": "success", "property_id": property_id, "env_key": env_key, "phones": phones} except Exception as e: return JSONResponse(status_code=500,content={"status": "error", "message": str(e)}) except Exception as e: logging.error(f"Error parsing {LOGGING_PREFIX}_{default_request.path_params}: {e}") raise HTTPException(500, detail=str(e))