from datetime import datetime,timedelta,timezone # import datetime import json import os import random from urllib.parse import unquote import firebase_admin import pyrebase from environment import Environment from exceptions import OfferNotFoundForUserException from firebase_admin import auth, firestore from services.dynamic_reward_service import DynamicRewardService from services.integrations.integrations_service import IntegrationsService from services.notifications.sms_service import SMSService from services.scheduling.scheduling_service import SchedulingService import requests import time from google.cloud.firestore_v1.base_query import FieldFilter import logging class FirestoreService: cred = Environment.FIREBASE_CREDENTIALS firebase_admin.initialize_app(cred) pyrebase = pyrebase.initialize_app(Environment.FIREBASE_CONFIG) auth = pyrebase.auth() firestore_client = firestore.client() FB_GRAPH_API_URL = f"https://graph.facebook.com/v19.0" app_id = '1154188842291843' app_secret = 'b79ab3ff9521b0e439a522eab93d71cc' def get_user_file(phone_number): doc_ref = FirestoreService.firestore_client.collection("users").document( phone_number ) doc_snapshot = doc_ref.get() if doc_snapshot.exists: return doc_snapshot else: return {} def get_autopilot_file(client_id): try: return ( FirestoreService.firestore_client.collection("autopilot") .document(client_id) .get() ) except: return {} def get_white_labeled_kits(domain_name): try: doc_ref = FirestoreService.firestore_client.collection("whiteLabeledKits").document(domain_name) doc = doc_ref.get() return doc.to_dict() if doc.exists else {} except Exception as e: print(f"Error fetching white labeled kits: {e}") return {} def get_offer_info(offer_id): doc_ref = FirestoreService.firestore_client.collection("offers").document( offer_id ) doc_snapshot = doc_ref.get() return doc_snapshot.to_dict() def get_client_info(client_id): doc_ref = FirestoreService.firestore_client.collection("clients").document( client_id ) doc_snapshot = doc_ref.get() return doc_snapshot.to_dict() def get_client_signups(client_id): doc_ref = FirestoreService.firestore_client.collection( "clientSignups" ).document(client_id) client_doc = doc_ref.get() if client_doc.exists: client_stats_dict = client_doc.to_dict() if "signups" in client_stats_dict: return client_stats_dict["signups"] else: client_stats_dict = {} return {} def get_all_influencers(): doc_ref = FirestoreService.firestore_client.collection( "influencers" ) if doc_ref: return # Adds an autopilot offer to a user's acceptedOffers and hits them with a text def send_offer_to_wallet(phone_number, offer_id, multiplier=1.0): user_file = FirestoreService.get_user_file(phone_number) if not user_file.exists: raise Exception("User file not found.") user_file = user_file.to_dict() accepted_offers = list(user_file["acceptedOffers"]) offer_info = FirestoreService.get_offer_info(offer_id) if ( offer_info.get("compensation") is None or offer_info.get("compensation") == 0 ) and (offer_info.get("minCompensation") and offer_info.get("maxCompensation")): # calculate compensation amount using dynamic reward service tailored_min_comp, tailored_max_comp = ( DynamicRewardService.calculate_tailored_compensation_bounds( offer_info["minCompensation"], offer_info["maxCompensation"], user_file["instagramFollowerCount"], multiplier=multiplier, ignore_reach=( offer_info["contentType"] == "ugc" or offer_info["contentType"] == "ugcPicture" or offer_info["contentType"] == "ugcVideo" or offer_info.get("isUgc") ), ) ) compensation = tailored_max_comp else: tailored_max_comp = 0 tailored_min_comp = 0 compensation = offer_info.get("compensation") # add offer to user's acceptedOffers accepted_offer = { "UTCTimeAccepted": round(datetime.datetime.utcnow().timestamp() * 1000), "UTCTimeRedeemed": 0, "clientID": str(offer_info["clientID"]), "exampleUGC": offer_info["exampleUGC"], "discountCode": "", "finalValidation": False, "iPosted": False, "initialValidation": False, "isExpired": False, "isRedeemed": False, "isRejected": False, "locationID": "0", "numToPost": 1, "offerID": str(offer_id), "ugcURL": "", "autoAdded": True, "compensation": compensation, # generate random acceptanceID to differentiate between multiple acceptances of the same offer "acceptanceID": "".join( random.choices( "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789", k=20, ) ), } # if offer is dynamic, add in multiplier. Otherwise, leave it out. if offer_info.get("minCompensation") and offer_info.get("maxCompensation"): accepted_offer["multiplier"] = multiplier accepted_offers.append(accepted_offer) # update user file FirestoreService.firestore_client.collection("users").document( phone_number ).update({"acceptedOffers": accepted_offers}) # send text to alert user client_info = FirestoreService.get_client_info(offer_info["clientID"]) SMSService.send_private_offer_alert_text( phone_number, client_info["accountName"], client_ig_handle=client_info["instagramHandle"], content_type=offer_info["contentType"], credit_type=offer_info["percentOrCashOff"], compensation=offer_info.get("compensation"), min_compensation=tailored_min_comp, max_compensation=tailored_max_comp, whitelabel_url=client_info.get("whiteLabelDomain"), hard_cash=offer_info.get("isCash"), ) def push_offer_to_manual_dashboard( phone_number, offer_id, required_tagged_username, client_id, location_id, ig_is_private, content_type, ugc_url, client_name, specific_product, acceptance_id, tiktok_handle, email ): # print out all params print("push_offer_to_manual_dashboard FB service" + email) user_file = ( FirestoreService.firestore_client.collection("users") .document(str(phone_number)) .get() ) print(f"User file: {user_file}") user_file = user_file.to_dict() print(f"User file dict: {user_file}") accepted_offers = list(user_file["acceptedOffers"]) offer_data = None for offer in accepted_offers: if offer["offerID"] == str(offer_id): offer_data = offer if not offer_data: print("Offer not found in user file.") raise OfferNotFoundForUserException() utc_time_accepted = offer_data["UTCTimeAccepted"] user_instagram_handle = user_file["instagramHandle"] file_name = ( str(phone_number) + "_" + str(offer_id) + "_" + str(random.randint(0, 9999999)) ) data_to_push = { "UTCTimeAccepted": utc_time_accepted, "clientInstagramHandle": required_tagged_username, "finalValidation": False, "instagramHandle": user_instagram_handle, "instagramIsPrivate": ig_is_private, "offerID": str(offer_id), "clientID": str(client_id), "locationID": str(location_id), "phoneNumber": str(phone_number), "requirements": [], "status": "pending", "contentType": content_type, "ugcUrl": ugc_url, "clientName": client_name, "specificProduct": specific_product, "tiktokHandle": tiktok_handle, "email": email } if acceptance_id: data_to_push["acceptanceID"] = acceptance_id doc_ref = FirestoreService.firestore_client.collection( "manuallyValidatedOffers" ).document(file_name) doc_ref.set(data_to_push) print("Offer pushed to manual dashboard.") def increment_number_offers_left(offer_id): offer_ref = FirestoreService.firestore_client.collection("offers").document( offer_id ) offer_data = offer_ref.get().to_dict() offer_data["numberOffersLeft"] += 1 offer_ref.update(offer_data) def create_dashboard_admin_auth_and_file(email, password, account_name): user = FirestoreService.auth.create_user_with_email_and_password( email, password ) # 2024-04-04 Evince_dev_python # Send verification mail to new seat created # FirestoreService.auth.send_email_verification(user.get('idToken')) # Do not send email asking to verify, just verify here access_user = firebase_admin.auth.get_user_by_email(email) updated_user = firebase_admin.auth.update_user( access_user.uid, email_verified=True ) print( f"Email verification for {email} status updated successfully: ", updated_user.email_verified, ) print("Successfully created auth account: " + email) user_specs = auth.verify_id_token(user["idToken"]) user_uid = user_specs["uid"] FirestoreService.firestore_client.collection("clientDashboardAdmin").document( user_uid ).create( { "email": email, "clientIDs": [], "userID": user_uid, "accountName": account_name, } ) print("Successfully created firebase file for dashboard account: " + email) def add_additional_seat(email, password, full_name, admin_id): # create auth account with email/password user = FirestoreService.auth.create_user_with_email_and_password( email, password ) # 2024-04-04 Evince_dev_python # Send verification mail to new seat created FirestoreService.auth.send_email_verification(user.get("idToken")) # create firestore doc in additionalSeats collection with adminID, email, and fullName. (Doc ID is email). FirestoreService.firestore_client.collection("additionalSeats").document( email ).create( { "adminID": admin_id, "email": email, "fullName": full_name, } ) # add the email to the additionalSeats array in the admin's file admin_file = ( FirestoreService.firestore_client.collection("clientDashboardAdmin") .document(admin_id) .get() .to_dict() ) # create the additionalSeats array if it doesn't exist if "additionalSeats" not in admin_file: admin_file["additionalSeats"] = [] admin_file["additionalSeats"].append(email) FirestoreService.firestore_client.collection("clientDashboardAdmin").document( admin_id ).update({"additionalSeats": admin_file["additionalSeats"]}) print("Successfully created additional seat for: " + email) # 2024-04-05 Evince_dev_python # delete addition seat def delete_additional_seat(email): # Delete addition seat record from firestore # Get user from firebase Authentication user = auth.get_user_by_email(email) # Get users additional seat details additional_seat = FirestoreService.firestore_client.collection( "additionalSeats" ).document(email) additional_seat_doc = additional_seat.get() admin_id = ( additional_seat_doc.get("adminID") if additional_seat_doc.exists else None ) # remove addition seat from clients table admin_file = FirestoreService.firestore_client.collection( "clientDashboardAdmin" ).document(admin_id) admin_file.update({"additionalSeats": firestore.ArrayRemove([email])}) # delete addition seat additional_seat.delete() # delete user auth.delete_user(user.uid) print("Successfully deleted additional seat for: " + email) # add a new file to firestore with the specified file_name and metadata def create_content_metadata_file(file_name, metadata): FirestoreService.firestore_client.collection("contentMetadata").document( file_name ).set(metadata) # 2024-04-05 Evince_dev_python # get batch code for offer def get_batch_code(offer_id): doc_ref = FirestoreService.firestore_client.collection("batchCodes").document(offer_id) doc_snapshot = doc_ref.get() data = doc_snapshot.to_dict() if data: codes = data.get('codes', []) for code in codes: if not code['isUsed']: code['isUsed'] = True doc_ref.update({'codes': codes}) return code['code'] print("No unused code found.") else: print("Offer ID not found or document is empty.") def update_final_verification( phone_number, offer_id, is_valid, ugc_url, content_type, score, email, acceptance_id=None, rejected_reason=None, ): print(f"Updating final verification for {phone_number}") print("UPDATE FINAL VERIFICATION EMAIL " + email) # try to decode (if coming from Flutter, it will be encoded) try: phone_number = unquote(phone_number) ugc_url = unquote(ugc_url) except Exception as e: print(e) user_file = ( FirestoreService.firestore_client.collection("users") .document(phone_number) .get() ) user_file = user_file.to_dict() accepted_offers = list(user_file["acceptedOffers"]) compensation_amount = 0 cash_increase = 0 discount_amount = 0 client_info = {} approved = False for offer in accepted_offers: # Ensure offer is the correct offer if offer["offerID"] == offer_id and ( offer.get("acceptanceID") == acceptance_id or acceptance_id == None ): if offer[ "finalValidation" ]: # if offer is already validated, stop searching and break break offer_info = FirestoreService.get_offer_info(offer_id) client_id = offer_info["clientID"] client_info = FirestoreService.get_client_info(client_id) whitelabel_url = client_info.get("whiteLabelDomain") white_labeled_kit = {} if whitelabel_url is not None: white_labeled_kit = FirestoreService.get_white_labeled_kits(whitelabel_url) business_content_approved_text = "" business_content_rejected_text = "" if "businessContentApprovedText" in white_labeled_kit and white_labeled_kit["businessContentApprovedText"] is not None: business_content_approved_text = white_labeled_kit["businessContentApprovedText"] if "businessContentRejectedText" in white_labeled_kit and white_labeled_kit["businessContentRejectedText"] is not None: business_content_rejected_text= white_labeled_kit["businessContentRejectedText"] if is_valid: # check if the the post was verified successfully offer["ugcURL"] = ( ugc_url # .replace("/", "%2F") # add the ugcURL to the offer, Because we encode the URL, we need to make the link valid by replacing the / with %2F ) offer["finalValidation"] = True # store current number of followers to maintain accurate stats (in case follower count changes in the future) offer["followers"] = user_file["instagramFollowerCount"] approved = True # set score to score if it's not None and it's an int, otherwise default to 80 (should only be used for force-approvals) score = ( score if score is not None and isinstance(score, int) else 80 ) discount_code = None # calculate compensation amount (can either be static or dynamic) if offer_info["compensation"]: # static reward compensation_amount = offer_info.get("compensation") # dynamic reward elif offer_info.get("minCompensation") and offer_info.get( "maxCompensation" ): compensation_amount = DynamicRewardService.calculate_rounded_dynamic_compensation_amount( offer_info.get("minCompensation"), offer_info.get("maxCompensation"), score, user_file.get("instagramFollowerCount"), multiplier=offer.get("multiplier"), ignore_reach=( offer_info.get("contentType") == "ugc" or offer_info.get("isUgc") or offer_info.get("contentType") == "ugcPicture" or offer_info.get("contentType") == "ugcVideo" ), ) # set compensation amount in offer (only necessary for dynamic rewards) offer["compensation"] = compensation_amount if offer_info["isCash"]: cash_increase = compensation_amount # add cash increase to user's cash balance print(f"Cash increase for offer {str(offer_id)} is {str(cash_increase)}") else: # is not a cash offer (schedule reminders to use credit) # Generate discount code via integration(if applicable) discount_type = ( "cash" if offer_info["percentOrCashOff"] == "Cash" else "percent" ) discount_amount = compensation_amount if offer_info.get( "integrations" ): # check if the offer has an integration if offer_info.get( "isCashBack", False ): # Checking if this is a cashback offer (shopify only) refund_id = IntegrationsService.initiate_cashback_offer( offer_info.get( "compensation" ), # compensation will be the cash back percent back on the order client_info["integrations"], email, ) if not refund_id: print(f"Could not find latest order for {email}") else: print("Cashback done with refund ID " + str(refund_id)) else: # Single use discount offer creation discount_code = ( IntegrationsService.create_single_use_discount_code( discount_amount, discount_type, offer_info["integrations"], client_info["integrations"], offer_info, ) ) # 2024-06-03 Evince_dev_python # Get batch code for offer elif offer_info.get("isUploadCodes", False): discount_code = FirestoreService.get_batch_code(offer_id) print("discount code from batch code -- ", discount_code) if discount_code: offer["discountCode"] = discount_code print(f"Discount code {discount_code} generated for offer {offer_id}") expiration_days_in_seconds = offer_info.get("expirationLimit" * 86400, 2592000) # defaults to 30 seconds delays_in_seconds = ( [expiration_days_in_seconds * .00001, expiration_days_in_seconds * .00003, expiration_days_in_seconds * .00007, expiration_days_in_seconds * .00009] #[10,30, 60, 90] if Environment.USE_TEST_MODE else [expiration_days_in_seconds * 0.1, expiration_days_in_seconds * 0.3, expiration_days_in_seconds * 0.7, expiration_days_in_seconds * 0.9] ) # in production, send 4 dynamic reminders of the total expiration time times .1, .3, .7, and .9 for delay in delays_in_seconds: SchedulingService.create_task( http_target=f"{Environment.FLASK_SERVER_URL}/send-sms/credit-reminder/", headers={ "Authorization": Environment.FLASK_API_KEY, "Content-Type": "application/json", }, body= { "phone_number": phone_number, "business_name": client_info["accountName"], "compensation": discount_amount, "credit_type": discount_type, "offer_id": offer_id, "whitelabel_url": client_info.get( "whiteLabelDomain" ), }, http_method="POST", from_now_in_seconds=delay, ) try: # alert user, send new offer is autopilot is enabled SMSService.send_post_approved_text( phone_number, whitelabel_url=whitelabel_url, discount_code=discount_code, business_content_approved_text = business_content_approved_text ) delay_in_seconds = ( 15 if Environment.USE_TEST_MODE else random.randint(79200, 93600) ) # delay between 22 and 26 hours autopilotData = FirestoreService.get_autopilot_file(client_id) hasAutopilotOffers = ( bool(autopilotData) and len(autopilotData.keys()) > 1 ) if hasAutopilotOffers: SchedulingService.create_task( http_target=f"{Environment.FLASK_SERVER_URL}/autopilot/send-targeted-offer/", headers={ "Authorization": Environment.FLASK_API_KEY, "Content-Type": "application/json", }, http_method="POST", body={ "phone_number": phone_number, "event": "content_verification", "client_id": client_id, "prev_offer_id": offer_id, "prev_multiplier": offer.get("multiplier"), }, from_now_in_seconds=delay_in_seconds, ) except: print("error autopilot did not have data") else: # story did not meet offer requirements offer["isRejected"] = True FirestoreService.increment_number_offers_left( offer_id ) # undo the decrement of number of offers left that occurred on "I Posted" SMSService.send_post_rejected_text( phone_number, rejected_reason=rejected_reason, is_ugc_offer=content_type == "ugc" or content_type == "ugcPicture" or content_type == "ugcVideo", business_content_rejected_text=business_content_rejected_text ) FirestoreService.firestore_client.collection("users").document( phone_number ).update( { "acceptedOffers": accepted_offers, "cashBalance": user_file["cashBalance"] + cash_increase, } ) break return approved def get_instagram_bots(self): return FirestoreService.firestore_client.collection("instagramBots").get() # Methods for measuring stats for clients, locations, and offers # Returns detailed list of client stats for a client, including location-specific and offer-specific stats def get_detailed_client_stats(client_id): from concurrent.futures import ThreadPoolExecutor client_doc = ( FirestoreService.firestore_client.collection("clients") .document(client_id) .get() ) offer_stats = {} location_stats = {} customer_group_stats = {} locations_quantity = 0 offers_quantity = 0 active_offers_quantity = 0 if client_doc.exists: client_doc = client_doc.to_dict() locations_quantity = len(client_doc["locations"]) def fetch_offer_data(acceptedOffer): nonlocal offers_quantity, active_offers_quantity # Declare as nonlocal offer_doc = ( FirestoreService.firestore_client.collection("offers") .document(str(acceptedOffer)) .get() ) if offer_doc.exists: offers_quantity += 1 offer_doc = offer_doc.to_dict() if offer_doc["isActive"]: active_offers_quantity += 1 # Use ThreadPoolExecutor to fetch offer data in parallel with ThreadPoolExecutor() as executor: executor.map(fetch_offer_data, client_doc["offers"]) else: return { "success": False, "error": "Client not found.", } # return error if client not found client_signups = 0 client_posts = 0 client_reach = 0 # note: offer reach only applies to offers that have been posted and verified. location reach includes all users who signed up for the location client_story_reach = 0 client_acceptances = 0 client_redemptions = 0 client_customer_list = ( {} ) # track map of phone numbers to list of offer ids completed by customer success = False client_signups_doc = ( FirestoreService.firestore_client.collection("clientSignups") .document(client_id) .get() ) if not client_signups_doc.exists: print("No stats document found. Returning basic client stats.") return { "success": True, "client_stats": { "posts": client_posts, "reach": client_reach, "storyReach": client_story_reach, "signups": client_signups, "redemptions": client_redemptions, "locationQty": locations_quantity, "offerQty": offers_quantity, "activeOfferQty": active_offers_quantity, "customerList": client_customer_list, "acceptances": client_acceptances, }, "location_stats": location_stats, "offer_stats": offer_stats, "customer_group_stats": customer_group_stats, } client_signups_doc = client_signups_doc.to_dict() # remove signups from list that are not registered in firebase auth new_signups_for_client = {} for phone_number, location_id in client_signups_doc["signups"].items(): try: auth.get_user_by_phone_number( phone_number ) # Check if phone number is registered in firebase auth print(f"Phone number {phone_number} is registered in firebase auth.") new_signups_for_client[phone_number] = location_id except auth.UserNotFoundError: print( f"Number: {phone_number} is not registered in firebase auth. Excluding from signups list." ) client_signups_doc["signups"] = new_signups_for_client client_signups = len(client_signups_doc["signups"]) def fetch_user_data(phone_number): nonlocal client_signups, client_posts, client_reach, client_story_reach, client_acceptances, client_redemptions location_id = client_signups_doc["signups"][phone_number] if location_id not in location_stats: location_stats[location_id] = { "signups": 0, "posts": 0, "reach": 0, # note: reach is the same as story reach for locations "acceptances": 0, "redemptions": 0, } location_stats[location_id]["signups"] += 1 # get doc from users collection user_doc = ( FirestoreService.firestore_client.collection("users") .document(phone_number) .get() ) if user_doc.exists: user_doc = user_doc.to_dict() if ( phone_number not in client_customer_list ): # add new customer to client_customer_list if not already there client_customer_list[phone_number] = { "acceptedOffers": [], "postedOffers": [], "redeemedOffers": [], "reach": user_doc["instagramFollowerCount"], } client_reach += user_doc[ "instagramFollowerCount" ] # add to total reach for client for acceptedOffer in user_doc["acceptedOffers"]: # convert client_id to string if it is an int acceptedOffer["clientID"] = str(acceptedOffer["clientID"]) # convert location_id to string if it is an int acceptedOffer["locationID"] = str(acceptedOffer["locationID"]) if ( acceptedOffer["clientID"] == client_id ): # check if the offer is for this client if ( acceptedOffer["offerID"] not in offer_stats ): # add new offer to offer_stats if not already there offer_stats[acceptedOffer["offerID"]] = { "posts": 0, "reach": 0, # note: reach is the same as story reach for offers "acceptances": 0, "redemptions": 0, } if ( acceptedOffer["locationID"] not in location_stats ): # add new location to location_stats if not already there location_stats[acceptedOffer["locationID"]] = { "signups": 0, "posts": 0, "reach": 0, "acceptances": 0, "redemptions": 0, } # increment acceptances for client, location, and offer client_acceptances += 1 location_stats[acceptedOffer["locationID"]]["acceptances"] += 1 offer_stats[acceptedOffer["offerID"]]["acceptances"] += 1 client_customer_list[phone_number]["acceptedOffers"].append( acceptedOffer["offerID"] ) # add offer to customer's acceptedOffers list if acceptedOffer[ "finalValidation" ]: # check if offer was validated (posted and approved) client_posts += 1 # increment client posts # add to total story reach for client, if a post from the customer has not already been counted if ( len(client_customer_list[phone_number]["postedOffers"]) == 0 ): client_story_reach += user_doc["instagramFollowerCount"] location_stats[acceptedOffer["locationID"]][ "posts" ] += 1 # increment location posts offer_stats[acceptedOffer["offerID"]][ "posts" ] += 1 # increment offer posts offer_stats[acceptedOffer["offerID"]]["reach"] += user_doc[ "instagramFollowerCount" ] # add to offer reach client_customer_list[phone_number]["postedOffers"].append( acceptedOffer["offerID"] ) # add offer to customer's postedOffers list if acceptedOffer[ "isRedeemed" ]: # check if offer was redeemed for credit client_redemptions += 1 location_stats[acceptedOffer["locationID"]][ "redemptions" ] += 1 offer_stats[acceptedOffer["offerID"]][ "redemptions" ] += 1 client_customer_list[phone_number][ "redeemedOffers" ].append( acceptedOffer["offerID"] ) # add offer to customer's redeemedOffers list location_stats[location_id]["reach"] += user_doc[ "instagramFollowerCount" ] # add to total reach for location else: print( f"User {phone_number} not found in users collection. Skipping user." ) # Use ThreadPoolExecutor to fetch user data in parallel with ThreadPoolExecutor() as executor: executor.map(fetch_user_data, client_signups_doc["signups"]) if ( "customerGroups" in client_doc and len(client_doc["customerGroups"]) > 0 ): # compile stats for each customer group (if client has any customer groups) for customer_group_id in client_doc["customerGroups"]: customer_group_stats[customer_group_id] = { "customerQtySignedUp": 0, "customerQtyTotal": 0, "posts": 0, "reach": 0, "acceptances": 0, "redemptions": 0, } # fetch customer group doc customer_group_doc = ( FirestoreService.firestore_client.collection("customerGroups") .document(customer_group_id) .get() ) # fetch customer group doc if customer_group_doc.exists: customer_group_doc = customer_group_doc.to_dict() for customer_phone_number in customer_group_doc["users"]: customer_group_stats[customer_group_id][ "customerQtyTotal" ] += 1 # add 1 to total customer qty (customer may or may not actually be signed up) - intended to account for manually added phone numbers try: customer_group_stats[customer_group_id]["posts"] += len( client_customer_list[customer_phone_number][ "postedOffers" ] ) customer_group_stats[customer_group_id][ "acceptances" ] += len( client_customer_list[customer_phone_number][ "acceptedOffers" ] ) customer_group_stats[customer_group_id][ "redemptions" ] += len( client_customer_list[customer_phone_number][ "redeemedOffers" ] ) customer_group_stats[customer_group_id][ "reach" ] += client_customer_list[customer_phone_number]["reach"] customer_group_stats[customer_group_id][ "customerQtySignedUp" ] += 1 except: print( f"Customer {customer_phone_number} not found in customer list. No stats to add for customer." ) client_stats = { "posts": client_posts, "reach": client_reach, "storyReach": client_story_reach, "signups": client_signups, "redemptions": client_redemptions, "acceptances": client_acceptances, "locationQty": locations_quantity, "offerQty": offers_quantity, "activeOfferQty": active_offers_quantity, "customerList": client_customer_list, } success = True return { "success": success, "client_stats": client_stats, "location_stats": location_stats, "offer_stats": offer_stats, "customer_group_stats": customer_group_stats, } # update fields for client_id doc in clientSignups collection def update_detailed_client_stats(client_id, stats): utc_timestamp = str(round(datetime.datetime.utcnow().timestamp() * 1000)) client_doc = ( FirestoreService.firestore_client.collection("clientStats") .document(client_id) .get() ) if client_doc.exists: client_doc = client_doc.to_dict() else: client_doc = {} # create copy but without customerList stats_to_upload = stats["client_stats"].copy() del stats_to_upload["customerList"] client_doc[utc_timestamp] = stats_to_upload FirestoreService.firestore_client.collection("clientStats").document( client_id ).set(client_doc) # update client community stats client_community_stats_doc = ( FirestoreService.firestore_client.collection("clientCommunityStats") .document(client_id) .get() ) if client_community_stats_doc.exists: client_community_stats_doc = client_community_stats_doc.to_dict() else: client_community_stats_doc = {} # set client community stats latest customer list, with customer phone numbers as field names for customer in stats["client_stats"]["customerList"]: client_community_stats_doc[customer] = stats["client_stats"][ "customerList" ][customer] FirestoreService.firestore_client.collection("clientCommunityStats").document( client_id ).set(client_community_stats_doc) for location in stats["location_stats"]: if location != "" and location != "0": location_doc = ( FirestoreService.firestore_client.collection("locationStats") .document(location) .get() ) if location_doc.exists: location_doc = location_doc.to_dict() else: location_doc = {} location_doc[utc_timestamp] = stats["location_stats"][location] FirestoreService.firestore_client.collection("locationStats").document( location ).set(location_doc) for offer in stats["offer_stats"]: offer_doc = ( FirestoreService.firestore_client.collection("offerStats") .document(offer) .get() ) if offer_doc.exists: offer_doc = offer_doc.to_dict() else: offer_doc = {} offer_doc[utc_timestamp] = stats["offer_stats"][offer] FirestoreService.firestore_client.collection("offerStats").document( offer ).set(offer_doc) for customer_group in stats["customer_group_stats"]: customer_group_doc = ( FirestoreService.firestore_client.collection("customerGroupStats") .document(customer_group) .get() ) if customer_group_doc.exists: customer_group_doc = customer_group_doc.to_dict() else: customer_group_doc = {} customer_group_doc[utc_timestamp] = stats["customer_group_stats"][ customer_group ] FirestoreService.firestore_client.collection("customerGroupStats").document( customer_group ).set(customer_group_doc) return stats def get_and_update_detailed_admin_account_stats(admin_id, admin_doc=None): admin_stats = { "clientQty": 0, "locationQty": 0, "offerQty": 0, "activeOfferQty": 0, "signups": 0, "reach": 0, "storyReach": 0, "posts": 0, "redemptions": 0, "acceptances": 0, } admin_community_stats = {} if admin_doc is None: admin_doc = ( firestore.client.collection("clientDashboardAdmin") .document(admin_id) .get() ) if admin_doc.exists: admin_doc = admin_doc.to_dict() # run stats update for each client on admin account for client_id in admin_doc["clientIDs"]: client_stats = FirestoreService.get_detailed_client_stats(client_id) if not client_stats["success"]: print(client_stats["error"]) continue # increment admin stats according to child clients admin_stats["clientQty"] += 1 admin_stats["locationQty"] += client_stats["client_stats"][ "locationQty" ] admin_stats["offerQty"] += client_stats["client_stats"]["offerQty"] admin_stats["activeOfferQty"] += client_stats["client_stats"][ "activeOfferQty" ] admin_stats["signups"] += client_stats["client_stats"]["signups"] admin_stats["reach"] += client_stats["client_stats"]["reach"] admin_stats["storyReach"] += client_stats["client_stats"]["storyReach"] admin_stats["posts"] += client_stats["client_stats"]["posts"] admin_stats["redemptions"] += client_stats["client_stats"][ "redemptions" ] admin_stats["acceptances"] += client_stats["client_stats"][ "acceptances" ] # add client community stats to admin community stats for customer in client_stats["client_stats"]["customerList"]: admin_community_stats[customer] = client_stats["client_stats"][ "customerList" ][customer] FirestoreService.update_detailed_client_stats( client_id, client_stats ) # update latest stats for client utcTimestamp = str( round(datetime.datetime.utcnow().timestamp() * 1000) ) # add timestamp as utc time in milliseconds, rounded to the nearest millisecond # update latest stats for admin account admin_stats_doc = ( FirestoreService.firestore_client.collection("adminStats") .document(admin_id) .get() ) if admin_stats_doc.exists: admin_stats_doc = admin_stats_doc.to_dict() else: admin_stats_doc = {} # set admin community stats latest customer list, with customer phone numbers as field names # overwrite adminCommunityStats doc with latest customer list FirestoreService.firestore_client.collection( "adminCommunityStats" ).document(admin_id).set(admin_community_stats) admin_stats_doc[utcTimestamp] = admin_stats FirestoreService.firestore_client.collection("adminStats").document( admin_id ).set(admin_stats_doc) return admin_stats # update stats for all admin accounts, including all child clients, locations, and offers def refresh_and_update_all_stats(): from concurrent.futures import ThreadPoolExecutor admin_docs = FirestoreService.firestore_client.collection( "clientDashboardAdmin" ).get() def worker(admin_doc): FirestoreService.get_and_update_detailed_admin_account_stats( admin_doc.id, admin_doc=admin_doc ) # Use ThreadPoolExecutor to run the workers in parallel with ThreadPoolExecutor() as executor: executor.map(worker, admin_docs) return True def update_metadata_stats(file_name, metadata): print( f"inside update metadata function. file_name: {file_name} metadata: {metadata}" ) FirestoreService.firestore_client.collection("contentMetadata").document( file_name ).update(metadata) # 2024-04-18 Evince_dev_python # Get offer details def get_offer_stats_details(offer_id): offer_details_with_stats = {} batchCodes = [] batchCodeDetails = [] offer_details = FirestoreService.firestore_client.collection( "offers").document(offer_id).get().to_dict() offer_stats = FirestoreService.firestore_client.collection( "offerStats").document(offer_id).get().to_dict() if offer_details: offer_details_with_stats.update(offer_details) offer_details_with_stats["stats"] = {"timestamps": offer_stats} # 2024-06-14 Evince_dev_python # batch code details if offer_details.get('isUploadCodes', False): batch_offer_details = FirestoreService.firestore_client.collection( "batchCodes").document(offer_id).get().to_dict() for batch_code in batch_offer_details.get('codes', []): if batch_code.get('isUsed') == False: batchCodes.append(batch_code.get('code')) batchCodeDetails.append(batch_code) offer_details_with_stats["batchCodes"] = batchCodes offer_details_with_stats["batchCodeDetails"] = batchCodeDetails return offer_details_with_stats def update_influencer_data(influencer_data): instagramId = influencer_data["instagramId"] try: influencer_doc_ref = FirestoreService.firestore_client.collection( "influencers").document(instagramId) if influencer_doc_ref.get().exists: influencer_doc_ref.update(influencer_data) print(f"Document {instagramId} updated successfully.") else: influencer_doc_ref.set(influencer_data) print(f"Creating new Document: {instagramId}") except: print("Error updating influencer: " + instagramId) # 2024-08-09 Evince_dev_python def check_facebook_token(access_token): """Check the validity of the Facebook token.""" print() # response = requests.get(f"{FirestoreService.FB_GRAPH_API_URL}/me", params={"access_token": access_token}) query = f"{FirestoreService.FB_GRAPH_API_URL}/me?fields=id%2Cname&access_token={access_token}" response = requests.get(query) # print("\ntoken---",response.json()) return response.json() # 2024-08-09 Evince_dev_python def generate_new_facebook_token(access_token, app_id, app_secret): """Refresh the Facebook long-lived token.""" response = requests.get(f"{FirestoreService.FB_GRAPH_API_URL}/oauth/access_token", params={ "grant_type": "fb_exchange_token", "client_id": app_id, "client_secret": app_secret, "fb_exchange_token": access_token }) return response.json() # 2024-08-22 Evince_dev_python def get_user_token_only(fbUserId): user_docs = FirestoreService.firestore_client.collection( 'metaInformation').where(filter=FieldFilter('fbUserId', '==', fbUserId)).limit(1).get() if user_docs: user_doc = user_docs[0] # Get the first document from the list user_data = user_doc.to_dict() # Convert the document to a dictionary meta = user_data.get("meta", {}) # Get the 'meta' dictionary, default to an empty dictionary if not found access_token = meta.get("accessToken") print("\naccess =", access_token) if access_token == '': return 'missing auth token' else: return access_token return 'user not found' # 2024-08-09 Evince_dev_python # Get user token etails def get_user_token_details(userId): user_doc = FirestoreService.firestore_client.collection( 'metaInformation').document(userId).get().to_dict() print("use_doc",user_doc) # return user_doc if user_doc is not None: user_data = user_doc meta = user_data.get("meta", {}) # Get the 'meta' dictionary, default to an empty dictionary if not found access_token = meta.get("accessToken") if access_token == '': return 'missing auth token' # refresh_token = user_data.get('refresh_token') # Check if the token is valid token_info = FirestoreService.check_facebook_token(access_token) if 'error' in token_info: new_token_info = FirestoreService.generate_new_facebook_token(access_token, FirestoreService.app_id, FirestoreService.app_secret) # print("new---==-----",new_token_info) seconds = user_doc['meta']['expiresIn'] future_time = datetime.now() + timedelta(seconds=seconds) # Convert future time to Unix time unix_time = int(future_time.timestamp()) # if 'error' in new_token_info: # return new_token_info # If token is invalid or expired, refresh it if 'access_token' in new_token_info: # Update the Firebase database with the new token in the 'meta' key FirestoreService.firestore_client.collection('metaInformation').document(userId).update({ 'meta.accessToken': new_token_info['access_token'], 'meta.expiresIn': new_token_info['expires_in'], 'meta.dataAccessExpirationTime':unix_time, 'meta.isTokenRefreshed': True }) user_doc = FirestoreService.firestore_client.collection( 'metaInformation').document(userId).get().to_dict() return {"data": user_doc,"status": "success", "message": "Token refreshed", "access_token": new_token_info['access_token']},201 else: return {"status": "error", "message": "Failed to refresh token"}, 422 # return {"status": "error", "message": "Need to refresh token"}, 401 else: # expire_time = user_doc['meta']['expiresIn'] data_access_time = user_doc['meta']['dataAccessExpirationTime'] current_unix_time = int(time.time()) remain_time = data_access_time - current_unix_time if remain_time < (15 * 60): if user_doc['meta']['isTokenRefreshed'] == False: new_token_info = FirestoreService.generate_new_facebook_token(access_token, FirestoreService.app_id, FirestoreService.app_secret) seconds = user_doc['meta']['expiresIn'] future_time = datetime.now() + timedelta(seconds=seconds) # Convert future time to Unix time unix_time = int(future_time.timestamp()) # If token is invalid or expired, refresh it if 'access_token' in new_token_info: # Update the Firebase database with the new token in the 'meta' key FirestoreService.firestore_client.collection('metaInformation').document(userId).update({ 'meta.accessToken': new_token_info['access_token'], 'meta.expiresIn': new_token_info['expires_in'], 'meta.dataAccessExpirationTime':unix_time, 'meta.isTokenRefreshed': True }) user_doc = FirestoreService.firestore_client.collection( 'metaInformation').document(userId).get().to_dict() return {"data": user_doc,"status": "success", "message": "Token refreshed", "access_token": new_token_info['access_token']},201 else: return {"status": "error", "message": "Failed to refresh token"}, 422 else: # Check if the token is valid token_info = FirestoreService.check_facebook_token(access_token) if 'error' in token_info: return {"status": "error", "message": "Token Expired..."}, 401 else: new_token_info = FirestoreService.generate_new_facebook_token(access_token, FirestoreService.app_id, FirestoreService.app_secret) seconds = user_doc['meta']['expiresIn'] future_time = datetime.now() + timedelta(seconds=seconds) # Convert future time to Unix time unix_time = int(future_time.timestamp()) # If token is invalid or expired, refresh it if 'access_token' in new_token_info: # Update the Firebase database with the new token in the 'meta' key FirestoreService.firestore_client.collection('metaInformation').document(userId).update({ 'meta.accessToken': new_token_info['access_token'], 'meta.expiresIn': new_token_info['expires_in'], 'meta.dataAccessExpirationTime':unix_time, 'meta.isTokenRefreshed': True }) user_doc = FirestoreService.firestore_client.collection( 'metaInformation').document(userId).get().to_dict() return {"data": user_doc,"status": "success", "message": "Token refreshed", "access_token": new_token_info['access_token']},201 else: return {"status": "error", "message": "Failed to refresh token"}, 422 return user_doc, 200 else: return {"status": "error", "message": "User not found"}, 404 # 2024-08-21 Evince_dev_python def send_direct_message(page_token,page_id,dm_template,sender_id,data_base_data,doc_id,dm_time): url = f"{FirestoreService.FB_GRAPH_API_URL}/{page_id}/messages" message = json.dumps({'text': dm_template}) sender_data= str({'id':sender_id}) payload = f'recipient={sender_data}&messaging_type=RESPONSE&message={message}&access_token={page_token}' headers = { 'Content-Type': 'application/x-www-form-urlencoded' } response = requests.request("POST", url, headers=headers, data=payload) ig_user_details=FirestoreService.get_ig_user_details(sender_id,page_token) if response.ok: # data_base_data['timestamp']=str(round(datetime.utcnow().timestamp() * 1000)) data_base_data['timestamp']=str(datetime.now().strftime('%d-%m-%Y %I:%M:%S %p')) data_base_data['message']=dm_template data_base_data['user']=ig_user_details['name'] data_base_data['username']=ig_user_details['username'] data_base_data['dm_time']=dm_time #instagram timestamp # Update the document by adding the new JSON under the "dmLogs" key FirestoreService.firestore_client.collection('metaInformation').document(doc_id).update({ "dmLogs": firestore.ArrayUnion([data_base_data]) }) return True else: return False def get_ig_user_details(user_ig_id,page_token): # fetch user's name, username and id from sender's instagram id url = f"{FirestoreService.FB_GRAPH_API_URL}/{user_ig_id}?fields=id,username,name&access_token={page_token}" response = requests.request("GET",url) if response.ok: res=response.json() if res['name'] != '' and res['username']!= '': res_data=res elif res['name'] != '' and res['username']=='': res_data={"name":res['name'], "username":""} elif res["username"] != '' and res['name']=='': res_data={"name":'',"username":res["username"]} elif res["name"]=='' and res["username"]!='': res_data={"name":res["id"], "username":res['username']} else: res_data={"name":user_ig_id, "username":""} else: res_data={"name":user_ig_id, "username":""} return res_data # 2024-08-21 Evince_dev_python @staticmethod def should_send_dm(client_id): # Fetch user documents from Firestore using 'filter' keyword argument print("dm function-=-=") user_docs = FirestoreService.firestore_client.collection('metaInformation').where(filter=FieldFilter('fbUserId', '==', client_id)).limit(1).get() print("\nuser_docs =", user_docs) for user_doc in user_docs: user_data = user_doc.to_dict() if user_data.get('dmEnable', False): return True return False # 2024-08-21 Evince_dev_python @staticmethod def get_dm_template(client_id): # Fetch user documents from Firestore using 'filter' keyword argument user_docs = FirestoreService.firestore_client.collection('metaInformation').where(filter=FieldFilter('fbUserId', '==', client_id)).limit(1).get() for user_doc in user_docs: user_data = user_doc.to_dict() return user_data.get('dmTemplate', 'Thank you for mentioning us! We appreciate your support.') return 'Thank you for mentioning us! We appreciate your support.' # 2024-09-18 Evince_dev_python # def get_all_access_tokens(): # access_tokens = [] # # Replace 'your_collection_name' with your actual collection name # docs = FirestoreService.firestore_client.collection('metaInformation').get() # print("docs",docs) # for doc in docs: # print('doc-=-',doc) # doc= doc.to_dict() # # Assuming the access token is stored under the 'access_token' field # if 'accessToken' in doc['meta']: # access_tokens.append(doc['meta']['accessToken']) # return access_tokens # def get_page_details_using_token(user_token): # url = f"{FirestoreService.FB_GRAPH_API_URL}/122109758918467036/accounts?access_token={user_token}" # print("url-",url) # response = requests.get(url) # return response # 2024-08-21 Evince_dev_python # add webhook response data and call dm api. def store_webhook_mention_response(mentions_webhook_data): # FirestoreService.firestore_client.collection('instagramMentions').add(data) logging.info(f"Webhook response stored in the db") for entry in mentions_webhook_data['entry']: if message_data := entry.get("messaging",[]): for data in message_data: instagram_user_id = data["recipient"]["id"] dm_time = entry['time'] sender_id = data["sender"]["id"] CDNUrl = data["message"]["attachments"][0]["payload"]["url"] data_base_data={"instagramUserId":instagram_user_id,"userId":sender_id,"CDNUrl":CDNUrl} user_docs = FirestoreService.firestore_client.collection('metaInformation').where(filter=FieldFilter('instagramId', '==', instagram_user_id)).limit(1).get() if user_docs: for doc in user_docs: # Document ID doc_id = doc.id page_token = doc.get('pageToken') page_id = doc.get('pageId') dm_template = doc.get('dmTemplate') dm_enable = doc.get('dmEnable') if dm_template == '': dm_template='Thank you for mentioning us! We appreciate your support. 😊' if dm_enable: messsage_func_response = FirestoreService.send_direct_message(page_token,page_id,dm_template,sender_id,data_base_data,doc_id,dm_time) print("\nmesssage_func_reson",messsage_func_response) return messsage_func_response else: dmenabled = "False" return dmenabled else: continue