find-object-bundle-builder/FindObjectBundleBuilder/Tools/firebase/firebase_helper.py

#!/usr/bin/env python
# coding:utf-8
import io
import json
import os
import sys
import requests

curr_dir = os.path.split(os.path.abspath(__file__))[0]
print(curr_dir)
sys.path.append(os.path.join(curr_dir, "../"))

import utils as utils
import config as config
import firebase_admin
from firebase_admin import credentials, db
from firebase_admin import storage, firestore
from oauth2client.service_account import ServiceAccountCredentials


GOOGLE_SERVER_FILE = os.path.join(curr_dir, "find-object-a08e5-f833603c7dd7.json")
REMOTE_CONFIG_FILE = os.path.join(curr_dir, "remote_config.json")
PROJECT_ID = "find-object-45a0d"
BASE_URL = "https://firebaseremoteconfig.googleapis.com"
REMOTE_CONFIG_ENDPOINT = "v1/projects/" + PROJECT_ID + "/remoteConfig"
REMOTE_CONFIG_URL = BASE_URL + "/" + REMOTE_CONFIG_ENDPOINT

def singleton(cls):
    _instance = {}

    def inner():
        if cls not in _instance:
            _instance[cls] = cls()
        return _instance[cls]
    return inner

@singleton
class FirebaseHelper:
    def __init__(self):
        self.firebase_app = None
        self.storage_instance = None
        self.config = {}
        self.config["project_id"] = PROJECT_ID
        self.init()

    def init(self):
        self.init_firebase(self.config["project_id"])
        self.init_storage(self.config["project_id"])

    def init_firebase(self, project_id):
        storage_bucket = "gs://" + project_id + ".firebasestorage.app"
        databaseURL = "https://" + project_id + ".firebaseio.com/"
        cred = credentials.Certificate(GOOGLE_SERVER_FILE)
        self.firebase_app = firebase_admin.initialize_app(cred, {
            "databaseURL": databaseURL,
            "storageBucket": storage_bucket
        }, name=project_id)

        refFirestore = firestore.client(app=self.firebase_app)
        print("初始 firebase 成功")

    def init_storage(self, project_id):
        bucket = project_id + ".firebasestorage.app"
        if self.firebase_app:
            self.storage_instance = storage.bucket(name=bucket, app=self.firebase_app)
            print("初始 storage 成功")

    def get_files(self, prefix=""):
        print("prefix = " + prefix)
        blobs = self.storage_instance.list_blobs(prefix=prefix)
        return blobs

    def get_files_match(self, prefix, match_glob):
        blobs = self.storage_instance.list_blobs(prefix=prefix, match_glob=match_glob)
        return blobs

    def get_files_all_versions(self, prefix=""):
        print("prefix = " + prefix)
        blobs = self.storage_instance.list_blobs(prefix=prefix, versions=True)
        return blobs

    def get_file(self, storage_file, generation=None):
        blob = self.storage_instance.get_blob(storage_file, generation=generation)
        return blob

    def upload_file_no_metadata(self, storage_file, file):
        try:
            upload_blob = self.storage_instance.blob(storage_file)
            if file.endswith(".json"):
                utils.gzip_file(file)
                upload_blob.content_encoding = "gzip"
                upload_blob.upload_from_filename(file + '.gz')
                os.unlink(file + '.gz')
            else:
                upload_blob.upload_from_filename(file)
            blob = self.storage_instance.get_blob(storage_file)
            print(file + "   上传成功")
            return blob.generation
        except Exception as e:
            raise Exception(e)

    def upload_file(self, storage_file, file, ext_meta={}):
        try:
            upload_blob = self.storage_instance.blob(storage_file)
            meta = {
                config.meta_hash: utils.calc_hash(file),
                config.meta_md5: utils.calc_md5(file),
            }

            meta.update(ext_meta)
            upload_blob.metadata = meta
            if file.endswith(".json"):
                utils.gzip_file(file)
                upload_blob.content_encoding = "gzip"
                upload_blob.upload_from_filename(file + '.gz')
                os.unlink(file + '.gz')
            else:
                upload_blob.upload_from_filename(file)
            blob = self.storage_instance.get_blob(storage_file)
            print(file + "   上传成功    generation = {}".format(blob.generation))
            return blob.generation
        except Exception as e:
            raise Exception(e)

    def get_access_token(self):
        file_value = ""
        with open(GOOGLE_SERVER_FILE, "r") as f:
            file_value = json.load(f)

        credentials = ServiceAccountCredentials.from_json_keyfile_dict(file_value, ["https://www.googleapis.com/auth/firebase.remoteconfig"])
        access_token_info = credentials.get_access_token()
        return access_token_info.access_token

    def get_remote_value(self):
        """
        获取RemoteConfig配置，并写入到remote_config.json中
        :return: ETag
        """
        try:
            headers = {
                "Authorization": "Bearer " + self.get_access_token()
            }
            resp = requests.get(REMOTE_CONFIG_URL, headers=headers)

            if resp.status_code == 200:
                with io.open(REMOTE_CONFIG_FILE, "wb") as f:
                    f.write(resp.text.encode("utf-8"))

                print("remote config 写入完成: remote_config.json")
                print("ETag from server: {}".format(resp.headers["ETag"]))
                return resp.headers["ETag"]
            else:
                print("remote_config.json获取失败")
                print(resp.text)
                return None
        except Exception as e:
            print("获取 RemoteConfig值失败   " + repr(e))
            raise Exception("Fail")

    def upload_remote_config_value(self, etag):
        """
        上传 remote_config.json 文件至firebase后台
        :param: etag
        """
        try:
            with open(REMOTE_CONFIG_FILE, "r", encoding="utf-8") as f:
                content = f.read()
            print("开始上传 remote config:>" + content + "<")
            headers = {
                "Authorization": "Bearer " + self.get_access_token(),
                "Content-Type": "application/json; UTF-8",
                "If-Match": etag
            }
            resp = requests.put(REMOTE_CONFIG_URL, data=content.encode("utf-8"), headers=headers)
            if resp.status_code == 200:
                print("推送成功")
                print("ETag from server: {}".format(resp.headers["ETag"]))
                return True
            else:
                print("推送失败")
                print(resp.text)
                return False
        except Exception as e:
            print("更新 RemoteConfig值失败   " + repr(e))
            return False

    def check_or_create_string_field(self, json_dict, keys, sub_value):
        arr = keys.split("/")
        field = json_dict
        index = 0
        for item in arr:
            if index == len(arr) - 1:
                print("item = " + item)
                field[item] = sub_value
                print(field[item])
            else:
                field = field[item]

            index = index + 1

    def check_or_create_json_field(self, json_dict, keys, sub_key, sub_value):
        arr = keys.split("/")
        field = json_dict
        index = 0
        for item in arr:
            if index == len(arr) - 1:
                print("item = " + item)
                value = json.loads(field[item])
                value[sub_key] = sub_value
                field[item] = json.dumps(value)
                print(field[item])
            else:
                field = field[item]

            index = index + 1

    def update_remote_config(self, group, condition, main_key, sub_key, value, is_string=False):
        try:
            etag = self.get_remote_value()
            online_txt = ""
            with open(REMOTE_CONFIG_FILE, "r") as f:
                online_txt = f.read()

            if online_txt != None and online_txt != "":
                online_json = json.loads(online_txt)
                keys = ""
                if condition is None:
                    keys = f"parameterGroups/{group}/parameters/{main_key}/defaultValue/value"
                else:
                    keys = f"parameterGroups/{group}/parameters/{main_key}/conditionalValues/{condition}/value"

                if is_string is False:
                    self.check_or_create_json_field(online_json, keys, sub_key, value)
                else:
                    self.check_or_create_string_field(online_json, keys, value)

                print("\n\n")
                print(online_json)
                print("\n\n")
                # 将online_json写入到remote_config.json文件
                utils.write_json(REMOTE_CONFIG_FILE, online_json)
                self.upload_remote_config_value(etag)
        except Exception as e:
            print(e)
            raise Exception("更新失败")

if __name__ == "__main__":
    helper = FirebaseHelper()
    blob1 = helper.get_file("Bundles/Android/Level/14cd688726c4d6b75b669990b16552e4.bundle")
    # blob2 = helper.get_file("Bundles/Android/Level/thesmiths/thesmiths_hcq20231212_1/a6d87302cb3f.bundle")
    print(blob1 is None)
    # print(blob2 is None)
    etag = helper.get_remote_value()
    print(etag)