PK!Kdophon_properties/__init__.py# encoding:utf-8 import sys import re from dophon_properties import tools __version__ = '0.1.0' LOGGER_ROOT = 'dophon_properties.' PROP_FUN = '.properties' MQ = { 'name': 'mq', 'modules': [ 'dophon_mq.properties', 'dophon_mq.properties.properties', 'dophon.mq.properties', 'dophon.mq.properties.properties' ] } DOPHON = { 'name': 'dophon', 'modules': [ 'properties', 'dophon.properties' ] } DB = { 'name': 'database', 'modules': [ 'dophon_db.properties', 'dophon_db.properties.properties', 'dophon.db.properties.properties', 'dophon.db.properties' ] } CLOUD_CLIENT = { 'name': 'cloud.client', 'modules': [ 'dophon.cloud.client.properties', 'dophon.cloud_client_properties', 'dophon_cloud.client.properties', 'dophon.cloud_client.properties', 'dophon_cloud_client.properties', ] } @tools.module_edge_print('properties module') def get_properties(prop_type: object, debug: bool = False): result = [get_property(per_prop_type,debug) for per_prop_type in prop_type] if isinstance(prop_type,list) else \ get_property(prop_type,debug) if isinstance(prop_type,dict) else None def get_property(prop_type: dict, debug: bool = False): m = __import__(LOGGER_ROOT + prop_type['name'] + PROP_FUN, fromlist=True) init_result = eval('m.init_properties()') for module_alias in prop_type['modules']: sys.modules[module_alias] = init_result if debug: # 打印模块信息 for name in dir(init_result): if not re.match('^__.+__$', name): print(name, '---', getattr(init_result, name)) PK!s dophon_properties/application.pyblueprint_path=['/routes'] db_pool_exe_time=False debug_trace=False error_info='HTML' host='127.0.0.1' ip_count=False mq={'remote_center': False} msg_queue_debug=False msg_queue_max_num=30 pool_conn_num=5 port=443 project_root='D:/Desktop/dophon-properties/dophon_properties' pydc_database='db' pydc_host='localhost' pydc_password='root' pydc_port=3306 pydc_user='root' pydc_xmlupdate_sech=False server_gevented=False server_processes=1 server_threaded=False ssl_context='adhoc' PK!#dophon_properties/cloud/__init__.pyPK!+B*dophon_properties/cloud/client/__init__.py# encoding:utf-8PK!>4dophon_properties/cloud/client/default_properties.py# encoding:utf-8 from ..utils import DictObject # 此处为微服务服务端配置 center_type = 'DOPHON' cloud_debug_trace = False client = DictObject({ 'center': ','.join(['http://127.0.0.1:8761/eureka']), 'host': '127.0.0.1', 'prefer_ip': False }) PK!zm,dophon_properties/cloud/client/properties.py# encoding:utf-8 """ 配置相关 """ import re from . import default_properties def init_properties(): try: properties = __import__('dophon.properties', fromlist=True) except: try: properties = __import__('application', fromlist=True) except: try: properties = __import__('config', fromlist=True) except: properties = default_properties finally: # 合成配置 for name in dir(default_properties): if not re.match('^__.+__$',name) and not hasattr(properties,name): setattr(properties,name,getattr(default_properties,name)) elif not re.match('^__.+__$',name): from ..utils import DictObject # 特别配置 def_dict_obj = getattr(default_properties,name) if isinstance(def_dict_obj,DictObject): # print(f'{name} - {def_dict_obj}') custom_dict = getattr(properties, name) for k,v in custom_dict.items(): def_dict_obj[k] = v setattr(properties,name,def_dict_obj) return properties PK!n)dophon_properties/cloud/utils/__init__.pyclass DictObject: def __init__(self, d: dict): for k, v in d.items(): self[k] = v def __setitem__(self, key, value): setattr(self, key, value)PK!+B&dophon_properties/database/__init__.py# encoding:utf-8PK!`yWW0dophon_properties/database/default_properties.py# encoding:utf-8 import os # 此处为数据库配置 project_root = os.getcwd() pool_conn_num = 5 # size of db connect pool pydc_host = 'localhost' pydc_port = 3306 pydc_user = 'root' pydc_password = 'root' pydc_database = 'db' pydc_xmlupdate_sech = False db_mode = 'ALL' # 数据库模式默认单例与分片兼容 db_pool_exe_time = False # 此处为分片数据库配置 db_cluster = [ # { # 'alias': 'a', # 'host': 'localhost', # 'port': 3306, # 'database': 'db', # 'user': 'root', # 'password': 'root' # } ] PK!P"(dophon_properties/database/properties.py# encoding:utf-8 """ 配置相关 """ import re from . import default_properties def init_properties(): try: properties = __import__('dophon.properties', fromlist=True) except: try: properties = __import__('application', fromlist=True) except: try: properties = __import__('config', fromlist=True) except: properties = default_properties finally: # 合成配置 for name in dir(default_properties): if not re.match('^__.+__$',name) and not hasattr(properties,name): setattr(properties,name,getattr(default_properties,name)) return properties PK!+B$dophon_properties/dophon/__init__.py# encoding:utf-8PK!;2.dophon_properties/dophon/default_properties.py# coding: utf-8 import os """ 配置集合 author:CallMeE date:2018-06-01 """ project_root = os.getcwd() # 服务器相关配置 server_threaded = False # 服务器多线程开关 server_gevented = False # 服务器gevent协程处理(会覆盖多线程开关) server_processes = 1 # 服务器多进程处理 debug_trace = False # 调试跟踪记录 # 此为开启ip防火墙模式(1秒不超过50次请求,60秒解冻) ip_count = False # 此处为服务器配置 host = '127.0.0.1' port = 443 ssl_context = 'adhoc' # 此处为路由文件夹配置 blueprint_path = ['/routes'] # route model dir path # 此处为日志配置 if debug_trace: logger_config = { # 'filename': 'app.log', # 'level': 'logging.DEBUG', 'format': '%(levelname)s : <%(module)s> (%(asctime)s) ==> %(filename)s {%(funcName)s} [line:%(lineno)d] ::: %(message)s', 'datefmt': '%Y-%m-%d %H:%M:%S' } error_info = 'HTML' PK!t΁&dophon_properties/dophon/properties.py# coding: utf-8 """ 配置管理 启动前尝试获得自定义配置(application.py) 无法获取则使用默认配置 """ import logging import os import re import sys from urllib3 import PoolManager from . import sys_utils from . import default_properties as def_prop from . import properties_handler # 取消flask banner os.environ.setdefault('WERKZEUG_RUN_MAIN', 'true') logger = logging.Logger(name=__name__) re_import_prop_flag = False properties_file_name_list = ['application.yml', 'application.yml', 'application.properties', 'application.prop', 'application.xml', 'application.json', 'application.py'] properties_file_handler = { 'py': properties_handler.py_handler, 'yml': properties_handler.yml_handler, # yml格式模块文件 'yaml': properties_handler.yml_handler, # yml格式模块文件 'properties': properties_handler.properties_handler, # yml格式模块文件 'prop': properties_handler.properties_handler, # yml格式模块文件 'xml': properties_handler.xml_handler, # xml格式模块文件 'json': properties_handler.json_handler # json格式模块文件 } default_properties = {} user_properties = {} class UnknownError(Exception): def __init__(self, *args, **kwargs): super(UnknownError, self).__init__(*args, **kwargs) self.__e_info = '错误: %s, 源头: %s' % args def __str__(self): return self.__e_info class NotFoundError(Exception): pass def read_self_prop(): global re_import_prop_flag, user_properties, default_properties try: u_prop = __import__('application', fromlist=True) # 记录原始字段 default_properties = sys_utils.to_dict(def_prop) user_properties = sys_utils.to_dict(u_prop) # 对比配置文件 for name in dir(def_prop): if re.match('__.*__', name): continue if name in dir(u_prop): continue setattr(u_prop, name, getattr(def_prop, name)) # 校验远程配置 if hasattr(u_prop, 'remote_prop'): print('获取远程配置', end='...') try: remote_prop_prop = getattr(u_prop, 'remote_prop') if 'mark' in remote_prop_prop and 'base' in remote_prop_prop: remote_prop_mark = remote_prop_prop['mark'] url = remote_prop_prop['base'] if isinstance(url, str) and isinstance(remote_prop_mark, list): url = (url if url.endswith('/') else (url + '/')) + '/'.join(remote_prop_mark) print('请求远程配置标签: %s' % (remote_prop_mark,), end='...') pool = PoolManager() try: res = pool.request( remote_prop_prop.get('method') if 'method' in remote_prop_prop else 'get', url) result = str(res.data, encoding='utf-8') if res.status == 404: # 非成功请求抛出异常 raise NotFoundError(result) try: print('应用远程配置') result = eval(result) if isinstance(result, dict): # 对比默认配置 # print(result) for name in dir(def_prop): if re.match('__.*__', name): continue if name in result: # print(name,'---',getattr(u_prop,name),'---',result.get(name)) setattr(u_prop, name, result.get(name)) # 遍历完毕删除远程地址 delattr(u_prop, 'remote_prop') else: raise TypeError('无法识别的配置类型') except Exception as eval_e: raise UnknownError(eval_e, result) except Exception as req_e: raise ConnectionError('远程配置获取失败,信息: %s' % req_e) else: raise TypeError('远程配置(remote_prop)参数值异常,获取远程配置失败') else: raise KeyError('缺少必要参数(base,mark)') except Exception as inner_e: logger.error(inner_e) return u_prop except Exception as e: get_properties = False # 添加项目路径到环境变量 print('添加项目路径到环境变量: %s' % (os.getcwd(),)) for (root, dirs, files) in os.walk(os.getcwd(), topdown=True): if get_properties: break for prop_file_name in properties_file_name_list: if get_properties: break if prop_file_name in files: file_info = prop_file_name.split('.') file_name = file_info[0] file_type = file_info[1] if file_type in properties_file_handler: properties_file_handler[file_type](root, prop_file_name, file_name, file_type) print('读取配置: %s%s%s' % (root, os.sep, prop_file_name,)) sys.path.append(root) else: logger.error('无法读取的配置文件类型(%s)' % (file_type,)) get_properties = True if not re_import_prop_flag: logger.info('重新引入配置') re_import_prop_flag = True return read_self_prop() else: logger.warning('重新引入配置失败') raise e def prop_init(): try: return read_self_prop() except Exception as e: logger.error(e) logger.error('没有找到自定义配置:(application.py)') logger.error('引用默认配置') logger.info('创建配置文件: application.py') with open(os.getcwd() + os.sep + 'application.py', 'wb') as app_prop: for name in dir(def_prop): if re.match('__.*__', name): continue else: value = getattr(def_prop, name) if type(value) is type(os): continue app_prop.write( bytes(name + '=' + re.sub('\\\\', '/', str(('\'' + value + '\'') if isinstance(value, str) else value)) + '\n', encoding='utf-8')) return read_self_prop() def init_properties(): return prop_init() PK!u^)).dophon_properties/dophon/properties_handler.py# -*- coding:utf-8 -*- import yaml import os from xml.parsers.expat import ParserCreate # yml文件格式后缀 yml_file_type = ['yml', 'yaml'] # properties 文件格式后缀 properties_file_type = ['properties', 'prop'] # xml 文件格式后缀 xml_file_type = ['xml'] # json 文件格式后缀 json_file_type = ['json'] # 配置文件基本标识 base_prop_key = [ 'project_root', 'server_threaded', 'server_gevented', 'server_processes', 'debug_trace', 'ip_count', 'host', 'port', 'ssl_context', 'blueprint_path', 'pool_conn_num', 'pydc_host', 'pydc_port', 'pydc_user', 'pydc_password', 'pydc_database', 'pydc_xmlupdate_sech', 'db_mode', 'db_pool_exe_time', 'msg_queue_max_num', 'msg_queue_debug', 'mq', 'logger_config', 'error_info' ] # 额外配置标识 extra_prop_key = ['remote_prop'] class Properties(object): """ properties / prop 文件反序列类 """ def __init__(self, file_path): self.file_path = file_path self.properties = {} def __getDict(self, strName, dictName, value): if (strName.find('.') > 0): k = strName.split('.')[0] dictName.setdefault(k, {}) return self.__getDict(strName[len(k) + 1:], dictName[k], value) else: try: dictName[strName] = int(value) except: try: dictName[strName] = eval(value) except: dictName[strName] = value return def get_properties(self): try: pro_file = open(self.file_path, 'Ur') for line in pro_file.readlines(): line = line.strip().replace('\n', '') if line.find("#") != -1: line = line[0:line.find('#')] if line.find('=') > 0: strs = line.split('=') strs[1] = line[len(strs[0]) + 1:] self.__getDict(strs[0].strip(), self.properties, strs[1].strip()) except Exception as e: raise e else: pro_file.close() return self.properties class Xml: """ xml文件反序列类 """ LIST_TAGS = ['COMMANDS'] def __init__(self, data=None): self._parser = ParserCreate() self._parser.StartElementHandler = self.start self._parser.EndElementHandler = self.end self._parser.CharacterDataHandler = self.data self.result = {} if data: self.feed(data) self.close() def feed(self, data): self._stack = [] self._data = '' self._parser.Parse(data, 0) def close(self): self._parser.Parse("", 1) del self._parser def start(self, tag, attrs): assert attrs == {} assert self._data.strip() == '' self._stack.append([tag]) self._data = '' def end(self, tag): last_tag = self._stack.pop() assert last_tag[0] == tag if len(last_tag) == 1: # leaf data = self._data else: if tag not in Xml.LIST_TAGS: # build a dict, repeating pairs get pushed into lists data = {} for k, v in last_tag[1:]: try: v = int(v) except: try: v = eval(v) except: v = v if k not in data: data[k] = v else: el = data[k] if type(el) is not list: data[k] = [el, v] else: el.append(v) else: # force into a list data = [{k: v} for k, v in last_tag[1:]] if self._stack: self._stack[-1].append((tag, data)) else: self.result = {tag: data} self._data = '' def data(self, data): self._data = data def get_dict(self): return self.result def prop_keys(): return base_prop_key + extra_prop_key def translate_to_file(temp_path, file_dict, file_name): """ 将其他形式配置文件翻译成py文件 :param temp_path: 源配置文件路径 :param file_dict: 配置文件内容字典(dict) :param file_name: 配置文件名 :return: """ with open(temp_path, 'w', encoding='utf-8') as d_prop_f: for k, v in file_dict.items(): if k in prop_keys(): d_prop_f.write(k + '=' + (('\'' + v + '\'') if isinstance(v, str) else str(v))) d_prop_f.write('\n') return file_name def py_handler(root, prop_file_name, file_name, file_type): """ py 文件转换处理 :return: """ # py文件不作处理 pass def yml_handler(root, prop_file_name, file_name, file_type): """ yml文件转换处理 :return: """ if file_type in yml_file_type: with open(root + os.sep + prop_file_name, 'r', encoding='utf-8') as f: file_dict = yaml.load(f) temp_path = root + os.sep + 'application.py' return translate_to_file(temp_path, file_dict, file_name) else: raise Exception('不能识别的yml或yaml类型') def properties_handler(root, prop_file_name, file_name, file_type): """ properties / prop 文件转换处理 :return: """ if file_type in properties_file_type: file_dict = Properties(root + os.sep + prop_file_name).get_properties() temp_path = root + os.sep + 'application.py' return translate_to_file(temp_path, file_dict, file_name) else: raise Exception('不能识别的properties或prop类型') def xml_handler(root, prop_file_name, file_name, file_type): """ xml文件转换处理 :return: """ if file_type in xml_file_type: with open(root + os.sep + prop_file_name, 'r', encoding='utf-8') as dom: file_dict = Xml(dom.read()).get_dict() temp_path = root + os.sep + 'application.py' return translate_to_file(temp_path, file_dict.get('config'), file_name) else: raise Exception('不能识别的xml类型') def json_handler(root, prop_file_name, file_name, file_type): """ json文件转换处理 :param root: :param prop_file_name: :param file_name: :param file_type: :return: """ if file_type in json_file_type: with open(root + os.sep + prop_file_name, 'r', encoding='utf-8') as j: import json file_dict = json.load(j) temp_path = root + os.sep + 'application.py' return translate_to_file(temp_path, file_dict, file_name) else: raise Exception('不能识别的json类型') PK!Z %dophon_properties/dophon/sys_utils.py# encoding:utf-8 import re def to_dict(obj: object) -> dict: result = {} for name in dir(obj): if re.match('__.*__', name): continue else: result[name] = getattr(obj, name) return result PK!+B dophon_properties/mq/__init__.py# encoding:utf-8PK!  *dophon_properties/mq/default_properties.py# encoding:utf-8 # 消息队列线程池工人数 msg_queue_max_num = 30 # 此处为消息队列配置 mq = { 'remote_center': False # 'remote_address': '127.0.0.1', # 'remote_port': 58800 } # 消息队列调试标识 msg_queue_debug = False PK!P""dophon_properties/mq/properties.py# encoding:utf-8 """ 配置相关 """ import re from . import default_properties def init_properties(): try: properties = __import__('dophon.properties', fromlist=True) except: try: properties = __import__('application', fromlist=True) except: try: properties = __import__('config', fromlist=True) except: properties = default_properties finally: # 合成配置 for name in dir(default_properties): if not re.match('^__.+__$',name) and not hasattr(properties,name): setattr(properties,name,getattr(default_properties,name)) return properties PK!maa#dophon_properties/tools/__init__.py# encoding:utf-8 import platform EXECUTE_FLAG = False def is_windows(): return 'Windows' == platform.system() def is_not_windows(): return not is_windows() def show_banner(): print(""" .o8 oooo "888 `888 .oooo888 .ooooo. oo.ooooo. 888 .oo. .ooooo. ooo. .oo. d88' `888 d88' `88b 888' `88b 888P"Y88b d88' `88b `888P"Y88b 888 888 888 888 888 888 888 888 888 888 888 888 888 888 888 888 888 888 888 888 888 888 888 888 `Y8bod88P" `Y8bod8P' 888bod8P' o888o o888o `Y8bod8P' o888o o888o 888 o888o Author:CallMeE Base:Flask Url:https://github.com/Ca11MeE/dophon.git https://gitee.com/callmee/dophon.git """) def module_edge_print(module_name): def fun(f): def fields(*args, **kwargs): global EXECUTE_FLAG if not EXECUTE_FLAG: print('------------------', module_name, '------------------\n') result = f(*args, **kwargs) if not EXECUTE_FLAG: print('\n------------------', module_name, '------------------\n\n\n') EXECUTE_FLAG = not EXECUTE_FLAG return result return fields return fun PK!H\TT-dophon_properties-0.1.2.post5.dist-info/WHEEL 1 0 нR \I$ơ7.ZON `h6oi14m,b4>4ɛpK>X;baP>PK!HSif0dophon_properties-0.1.2.post5.dist-info/METADATAJ1y]BP`E>t}mfob2wɲ7r l1_h=ׁS^Bp/P<2 0ͷѧ䉣+pnWH`t3*6bRCVwWz&VR.Qn~?o^ƍ8PK!H'e.dophon_properties-0.1.2.post5.dist-info/RECORDɒXE-`Ɍ^ ( L2P"|}gWwgdVor A."P`MX$X٢ˤ cbKH Y+SHixGjuu 8*TxSBǫ;ցՒT3ġ)pm7,=3I+yu{ qp7.t8v0oFwp֟H[!pZ.VPdyo%Q{8 NpUa < mƩ'?ܞޅt=##HETЗ2W-\i{{^;0dO%*yg;7>ɤ7wcXӪkG·xfA(>:>_V/]65 NQG9یO(| llE1lYoYS=Udo@f/.hF!ێd"-y#Cu{"e@1z,H>c/<-,#lx, l鞪= V09i λQJ?z8aEOpw {*=rwҍ I՚P$c:/( ]ؒ-MR/A# >kqZ#jܾwt%NaDSzS$؛f7\R`8 4{gt|hňA'RK:t`aB-čWĊ\uG*)ztUY?y,J*2IO⨤]kt&""TTWz b,?qY舄8I &λȦYw`Qμ [SQ--a؏?PK!Kdophon_properties/__init__.pyPK!s 7dophon_properties/application.pyPK!#T dophon_properties/cloud/__init__.pyPK!+B* dophon_properties/cloud/client/__init__.pyPK!>4 dophon_properties/cloud/client/default_properties.pyPK!zm,S dophon_properties/cloud/client/properties.pyPK!n)dophon_properties/cloud/utils/__init__.pyPK!+B&dophon_properties/database/__init__.pyPK!`yWW0dophon_properties/database/default_properties.pyPK!P"(ydophon_properties/database/properties.pyPK!+B$dophon_properties/dophon/__init__.pyPK!;2.dophon_properties/dophon/default_properties.pyPK!t΁&dophon_properties/dophon/properties.pyPK!u^)).9dophon_properties/dophon/properties_handler.pyPK!Z %xUdophon_properties/dophon/sys_utils.pyPK!+B Vdophon_properties/mq/__init__.pyPK!  *Wdophon_properties/mq/default_properties.pyPK!P""WXdophon_properties/mq/properties.pyPK!maa#l[dophon_properties/tools/__init__.pyPK!H\TT-bdophon_properties-0.1.2.post5.dist-info/WHEELPK!HSif0bdophon_properties-0.1.2.post5.dist-info/METADATAPK!H'e.cdophon_properties-0.1.2.post5.dist-info/RECORDPKjvg