#!/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): ## sub_key 该云控里面的json的某个参数的key try: etag = self.get_remote_value() if etag is None: print("获取 ETag 失败，无法更新 Remote Config") return False 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 group is not None and group != "": keys = f"parameterGroups/{group}/parameters/" else: keys = f"parameters/" if condition is not None and condition != "": keys = keys + f"{main_key}/conditionalValues/{condition}/value" else: keys = keys + f"{main_key}/defaultValue/value" # 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) # 检查上传结果 upload_success = self.upload_remote_config_value(etag) if upload_success: print(f"Remote Config 更新成功: {main_key}/{sub_key}") return True else: print(f"Remote Config 更新失败: {main_key}/{sub_key}") return False except Exception as e: print(f"更新 Remote Config 异常: {repr(e)}") return False 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)