import os import pymongo import datetime import time from .cloudService import GetConfig local_path = os.path.dirname(__file__) class DBUtils(GetConfig): """ 从安全的角度出发,将一些默认配置文件上传至COS中,接下来使用COS和它的子类的时候,在第一次使用时需要输入Cuny给的id和key 用于连接数据库等对象 当然,在db_default_download = False的时候,如果在运行路径下已经有配置文件了, 那么就不用再次下载了,也不用输入id和key 事实上这只需要运行一次,因为配置文件将会被下载至源码文件夹中 如果要自定义路径,请在继承的子类中编写__init__函数,将service_path定向到指定路径 """ __BASE_DIR: dict = None __PARAMS_DIR: dict = None db_base_path: str = f"{local_path}/conf/base_config.json" db_params_path: str = f"{local_path}/conf/params.json" db_default_download: bool = False @property def base_config(self): if self.__BASE_DIR is None: self.__BASE_DIR = self.load_json(self.db_base_path, self.db_default_download) return self.__BASE_DIR @property def db_config(self): return self.base_config["database_config"] @property def params_config(self): if self.__PARAMS_DIR is None: self.__PARAMS_DIR = self.load_json(self.db_params_path, self.db_default_download) return self.__PARAMS_DIR @property def size_dir(self): return self.params_config["size_config"] @property def func_dir(self): return self.params_config["func_config"] @property def wx_config(self): return self.base_config["wx_config"] def get_dbClient(self): return pymongo.MongoClient(self.db_config["connect_url"]) @staticmethod def get_time(yyyymmdd=None, delta_date=0): """ 给出当前的时间 :param yyyymmdd: 以yyyymmdd给出的日期时间 :param delta_date: 获取减去delta_day后的时间，默认为0就是当天 时间格式:yyyy_mm_dd """ if yyyymmdd is None: now_time = (datetime.datetime.now() - datetime.timedelta(delta_date)).strftime("%Y-%m-%d") return now_time # 输入了yyyymmdd的数据和delta_date,通过这两个数据返回距离yyyymmdd delta_date天的时间 pre_time = datetime.datetime(int(yyyymmdd[0:4]), int(yyyymmdd[4:6]), int(yyyymmdd[6:8])) return (pre_time - datetime.timedelta(delta_date)).strftime("%Y-%m-%d") # 获得时间戳 def get_timestamp(self, date_time:str=None) -> int: """ 输入的日期形式为:"2021-11-29 16:39:45.999" 真正必须输入的是前十个字符,及精确到日期,后面的时间可以不输入,不输入则默认置零 """ def standardDateTime(dt:str) -> str: """ 规范化时间字符串 """ if len(dt) < 10: raise ValueError("你必须至少输入准确到天的日期!比如:2021-11-29") elif len(dt) == 10: return dt + " 00:00:00.0" else: try: date, time = dt.split(" ") except ValueError: raise ValueError("你只能也必须在日期与具体时间之间增加一个空格,其他地方不能出现空格!") while len(time) < 10: if len(time) in (2, 5): time += ":" elif len(time) == 8: time += "." else: time += "0" return date + " " + time if date_time is None: # 默认返回当前时间(str), date_time精确到毫秒 date_time = datetime.datetime.now() # 转换成时间戳 else: date_time = standardDateTime(dt=date_time) date_time = datetime.datetime.strptime(date_time, "%Y-%m-%d %H:%M:%S.%f") timestamp_ms = int(time.mktime(date_time.timetuple()) * 1000.0 + date_time.microsecond / 1000.0) return timestamp_ms @staticmethod def get_standardTime(yyyy_mm_dd: str): return yyyy_mm_dd[0:4] + yyyy_mm_dd[5:7] + yyyy_mm_dd[8:10] def find_oneDay_data(self, db_name: str, collection_name: str, date: str = None) -> dict: """ 获取指定天数的数据，如果date is None，就自动寻找距今最近的有数据的那一天的数据 """ df = None # 应该被返回的数据 collection = self.get_dbClient()[db_name][collection_name] if date is None: # 自动寻找前几天的数据,最多三十天 for delta_date in range(1, 31): date_yyyymmdd = self.get_standardTime(self.get_time(delta_date=delta_date)) filter_ = {"date": date_yyyymmdd} df = collection.find_one(filter=filter_) if df is not None: del df["_id"] break else: filter_ = {"date": date} df = collection.find_one(filter=filter_) if df is not None: del df["_id"] return df def find_daysData_byPeriod(self, date_period: tuple, db_name: str, col_name: str): # 给出一个指定的范围日期，返回相应的数据(日期的两头都会被寻找) # 这个函数我们默认数据库中的数据是连续的，即不会出现在 20211221 到 20211229 之间有一天没有数据的情况 if len(date_period) != 2: raise ValueError("date_period数据结构：(开始日期，截止日期)") start, end = date_period # yyyymmdd delta_date = int(end) - int(start) if delta_date < 0: raise ValueError("传入的日期有误！") collection = self.get_dbClient()[db_name][col_name] date = start while int(date) <= int(end): yield collection.find_one(filter={"date": date}) date = self.get_standardTime(self.get_time(date, -1)) @staticmethod def find_biggest_valueDict(dict_: dict): # 寻找字典中数值最大的字段，要求输入的字典的字段值全为数字 while len(dict_) > 0: max_value = 0 p = None for key in dict_: if dict_[key] > max_value: p = key max_value = dict_[key] yield p, max_value del dict_[p] def copy_andAdd_dict(self, dict_base, dict_): # 深度拷贝字典，将后者赋值给前者 # 如果后者的键名在前者已经存在，则直接相加。这就要求两者的数据是数值型 for key in dict_: if key not in dict_base: dict_base[key] = dict_[key] else: if isinstance(dict_[key], int) or isinstance(dict_[key], float): dict_base[key] = round(dict_[key] + dict_base[key], 2) else: dict_base[key] = self.copy_andAdd_dict(dict_base[key], dict_[key]) return dict_base @staticmethod def compare_data(dict1: dict, dict2: dict, suffix: str, save: int, **kwargs): """ 有两个字典,并且通过kwargs会传输一个新的字典,根据字典中的键值我们进行比对,处理成相应的数据格式 并且在dict1中,生成一个新的键值,为kwargs中的元素+suffix save:保留几位小数 """ new_dict = dict1.copy() for key in kwargs: try: if kwargs[key] not in dict2 or int(dict2[kwargs[key]]) == -1 or float(dict1[kwargs[key]]) <= 0.0: # 数据不存在 data_new = 5002 else: try: data_new = round( ((float(dict1[kwargs[key]]) - float(dict2[kwargs[key]])) / float(dict2[kwargs[key]])) * 100 , save) except ZeroDivisionError: data_new = 5002 if data_new == 0.0: data_new = 0 except TypeError as e: print(e) data_new = 5002 # 如果没有之前的数据,默认返回0 new_dict[kwargs[key] + suffix] = data_new return new_dict @staticmethod def sum_dictList_byKey(dictList: list, **kwargs) -> dict: """ 有一个列表,列表中的元素为字典,并且所有字典都有一个键值为key的字段,字段值为数字 我们将每一个字典的key字段提取后相加,得到该字段值之和. """ sum_num = {} if kwargs is None: raise ImportError("Please input at least ONE key") for key in kwargs: sum_num[kwargs[key]] = 0 for dict_ in dictList: if not isinstance(dict_, dict): raise TypeError("object is not DICT!") for key in kwargs: sum_num[kwargs[key]] += dict_[kwargs[key]] return sum_num @staticmethod def sum_2ListDict(list_dict1: list, list_dict2: list, key_name, data_name): """ 有两个列表,列表内的元素为字典,我们根据key所对应的键值寻找列表中键值相同的两个元素,将他们的data对应的键值相加 生成新的列表字典(其余键值被删除) key仅在一个列表中存在,则直接加入新的列表字典 """ sum_list = [] def find_sameKey(kn, key_, ld: list) -> int: for dic_ in ld: if dic_[kn] == key_: post_ = ld.index(dic_) return post_ return -1 for dic in list_dict1: key = dic[key_name] # 键名 post = find_sameKey(key_name, key, list_dict2) # 在list2中寻找相同的位置 data = dic[data_name] + list_dict2[post][data_name] if post != -1 else dic[data_name] sum_list.append({key_name: key, data_name: data}) return sum_list @staticmethod def find_biggest_dictList(dictList: list, key: str = "key", data: str = "value"): """ 有一个列表，里面每一个元素都是一个字典 这些字典有一些共通性质，那就是里面都有一个key键名和一个data键名，后者的键值必须是数字 我们根据data键值的大小进行生成，每一次返回列表中data键值最大的数和它的key键值 """ while len(dictList) > 0: point = 0 biggest_num = int(dictList[0][data]) biggest_key = dictList[0][key] for i in range(len(dictList)): num = int(dictList[i][data]) if num > biggest_num: point = i biggest_num = int(dictList[i][data]) biggest_key = dictList[i][key] yield str(biggest_key), biggest_num del dictList[point] def get_share_data(self, date_yyyymmdd: str): # 获得用户界面情况 visitPage = self.find_oneDay_data(date=date_yyyymmdd, db_name="cuny-user-analysis", collection_name="daily-userVisitPage") if visitPage is not None: # 这一部分没有得到数据是可以容忍的.不用抛出模态框错误 # 获得昨日用户分享情况 sum_num = self.sum_dictList_byKey(dictList=visitPage["data_list"], key1="page_share_pv", key2="page_share_uv") else: # 此时将分享次数等置为-1 sum_num = {"page_share_pv": -1, "page_share_uv": -1} return sum_num @staticmethod def compare_date(date1_yyyymmdd: str, date2_yyyymmdd: str): # 如果date1是date2的昨天，那么就返回True date1 = int(date1_yyyymmdd) date2 = int(date2_yyyymmdd) return True if date2 - date1 == 1 else False def change_time(self, date_yyyymmdd: str, mode: int): # 将yyyymmdd的数据分开为相应的数据形式 if mode == 1: if self.compare_date(date_yyyymmdd, self.get_standardTime(self.get_time(delta_date=0))) is False: return date_yyyymmdd[0:4] + "年" + date_yyyymmdd[4:6] + "月" + date_yyyymmdd[6:8] + "日" else: return "昨日" elif mode == 2: date = date_yyyymmdd[0:4] + "." + date_yyyymmdd[4:6] + "." + date_yyyymmdd[6:8] if self.compare_date(date_yyyymmdd, self.get_standardTime(self.get_time(delta_date=0))) is True: return date + "~" + date + " | 昨日" else: return date + "~" + date @staticmethod def changeList_dict2List_list(dl: list, order: list): """ 列表内是一个个字典,本函数将字典拆解,以order的形式排列键值为列表 考虑到一些格式的问题,这里我采用生成器的形式封装 """ for dic in dl: # dic是列表内的字典元素 tmp = [] for key_name in order: key = dic[key_name] tmp.append(key) yield tmp def dict_mapping(self, dict_name: str, id_: str): """ 进行字典映射，输入字典名称和键名，返回具体的键值 如果不存在，则原路返回键名 """ try: return getattr(self, dict_name)[id_] except KeyError: return id_ except AttributeError: print(f"[WARNING]: 本对象内部不存在{dict_name}!") return id_ @staticmethod def dictAddKey(dic: dict, dic_tmp: dict, **kwargs): """ 往字典中加入参数，可迭代 """ for key in kwargs: dic[key] = dic_tmp[key] return dic if __name__ == "__main__": dbu = DBUtils()