PK!\Җoocjk_commons/__init__.py# -*- coding: utf-8 -*- # noinspection PyUnresolvedReferences from cjk_commons.__about__ import __version__ PK!$cjk_commons/logging_.py# -*- coding: utf-8 -*- # noinspection PyUnresolvedReferences import logging from pathlib import Path def add_logging_arguments(parser) -> None: parser.add_argument( '-l', '--level', nargs='?', help='Set log\'s level' ) parser.add_argument( '--log-file', nargs='?', help='Set log file' ) parser.add_argument( '--log-file-level', nargs='?', help='Set log file\'s level' ) def add_loggers(args, main_logger: logging.Logger, log_file_name: str = '') -> None: # noinspection PyUnresolvedReferences formatter: logging.Formatter = logging.Formatter( '[%(asctime)s,%(msecs)03d][%(name)s:%(lineno)d][%(levelname)s] %(message)s', datefmt='%y-%m-%d %H:%M:%S') if args.level is not None: level_str = args.level else: level_str = 'INFO' level_str = level_str.upper() level_int = getattr(logging, level_str, None) if not isinstance(level_int, int): raise ValueError('Invalid log level \'{0}\''.format(level_str)) # noinspection PyUnresolvedReferences ch: logging.StreamHandler = logging.StreamHandler() ch.setLevel(level_int) ch.setFormatter(formatter) main_logger.addHandler(ch) if args.log_file is not None: log_file_path = Path(args.log_file) if log_file_path.is_dir(): if Path(log_file_name).stem == log_file_name: log_file_name += '.log' log_file_path = Path(log_file_path, log_file_name) if args.log_file_level is not None: log_file_level_str = args.log_file_level else: log_file_level_str = 'INFO' log_file_level_str = log_file_level_str.upper() log_file_level_int = getattr(logging, log_file_level_str, None) if not isinstance(log_file_level_int, int): raise ValueError('Invalid log file level \'{0}\''.format(log_file_level_str)) # noinspection PyUnresolvedReferences fh: logging.FileHandler = logging.FileHandler(log_file_path) fh.setLevel(log_file_level_int) fh.setFormatter(formatter) main_logger.addHandler(fh) PK!Zcjk_commons/settings.py# -*- coding: utf-8 -*- from collections import OrderedDict from pathlib import Path from typing import Any from appdirs import site_data_dir, user_data_dir import yaml import yodl def get_attribute( kwargs: dict, kwargs_key: str, settings: OrderedDict = None, settings_key: str = None, default: Any = None, type_: type = str, allow_none: bool = False) -> Any: result = None type__ = type_ if default is not None: type__ = type(default) if kwargs_key in kwargs: result = kwargs[kwargs_key] if not isinstance(result, type__) and default is not None: raise TypeError('{0}'.format(kwargs_key) if kwargs_key else None) if settings is not None and settings_key is not None: if settings_key in settings: result = settings[settings_key] if not isinstance(result, type__) and default is not None: raise TypeError('{0}'.format(settings_key) if settings_key else None) if result is None: result = default if result is None and not allow_none: raise AttributeError('{0}'.format(settings_key)) return result def get_path_attribute( kwargs: dict, kwargs_key: str, settings: OrderedDict = None, settings_key: str = None, default_path: Path = None, is_dir: bool = True, check_if_exists: bool = True, create_dir: bool = True, create_parents: bool = True) -> Path: result = None if kwargs_key in kwargs: result = kwargs[kwargs_key] if not isinstance(result, Path): raise TypeError('{0}'.format(kwargs_key) if kwargs_key else None) elif settings is not None and settings_key is not None: if settings_key in settings: result_str = settings[settings_key] if not isinstance(result_str, str): raise TypeError('{0}'.format(settings_key) if settings_key else None) result = Path(result_str) if not isinstance(result, Path): raise TypeError('{0}'.format(settings_key) if settings_key else None) if result is None and isinstance(default_path, Path): result = default_path if result is None: raise AttributeError('{0}'.format(settings_key)) if result.exists(): if not is_dir and result.is_dir(): raise FileExistsError('{0} Not A File'.format(kwargs_key) if kwargs_key else None, result) elif is_dir and result.is_file(): raise NotADirectoryError('{0}'.format(kwargs_key) if kwargs_key else None, result) else: if check_if_exists: raise FileExistsError('{0}'.format(kwargs_key) if kwargs_key else None, result) if is_dir and create_dir: result.mkdir(parents=create_parents) return result class SettingsError(Exception): """Settings Error""" def get_settings(file_path=Path('settings.yaml'), **kwargs) -> OrderedDict: if not file_path.is_file(): app_name = get_attribute(kwargs, 'app_name', allow_none=True) app_author = get_attribute(kwargs, 'app_author', allow_none=True) file_path = Path(user_data_dir(app_name, app_author, roaming=True), file_path.name) if not file_path.is_file(): file_path = Path(site_data_dir(app_name, app_author), file_path.name) if file_path.is_file(): with file_path.open(encoding='utf-8') as settings_file: settings = yaml.load(settings_file, yodl.OrderedDictYAMLLoader) if settings is None: settings = OrderedDict() else: settings = OrderedDict() return settings class OrderedDictMergeException(Exception): """Ordered Dict Merge Exception""" def merge(a: dict, b: dict, path=None) -> dict: if path is None: path = [] for key in b: if key in a: if isinstance(a[key], dict) and isinstance(b[key], dict): merge(a[key], b[key], path + [str(key)]) elif a[key] == b[key]: pass else: raise OrderedDictMergeException('Conflict at \'{0}\''.format('.'.join(path + [str(key)]))) else: a[key] = b[key] return a PK!#Ocjk_commons/zip.py# -*- coding: utf-8 -*- import os from pathlib import Path import time from typing import List import zipfile def extract_from_zip(zip_path: Path, dir_path: Path) -> None: with zipfile.ZipFile(zip_path) as zip_file: for zip_member in zip_file.infolist(): zip_file.extract(zip_member, dir_path) zip_member_time = time.mktime(zip_member.date_time + (0, 0, -1)) os.utime(Path(dir_path, zip_member.filename), (zip_member_time, zip_member_time)) def write_to_zip(zip_path: Path, in_path: Path, file_paths: List[Path] = None) -> float: with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zip_file: mtime = -1. if in_path.is_file(): zip_file.write(in_path, in_path.name) mtime = in_path.stat().st_mtime elif in_path.is_dir(): if not file_paths: file_paths = [] for root, dirnames, filenames in os.walk(str(in_path)): for filename in filenames: file_paths.append(Path(root, filename)) for file_path in file_paths: if file_path.is_file(): zip_file.write(file_path, file_path.relative_to(in_path)) file_stat_result = file_path.stat() # todo if mtime < file_stat_result.st_mtime: mtime = file_stat_result.st_mtime return mtime PK!HڽTU!cjk_commons-3.3.3.dist-info/WHEEL A н#Z;/"d&F[xzw@Zpy3Fv]\fi4WZ^EgM_-]#0(q7PK!HO"Eq$cjk_commons-3.3.3.dist-info/METADATAKO@ԄNAP ˱y@ｃQBMgw=˙c#˕LD3& %5ΔJKR|*%ZSb m6MV3l3xCW&8m샫> /:\^yn 9«.=ڎF%p[A1-IS#1X:9ds2@~ֻa9&Cڡ5=l)7,3nXI&L+ £dU&EX4u~ñRgAXg~t\;^r~˭ S 9I 68% !\i=E(}#n~9Ef~YjE# V'3age'+H*s ɼ[D&6wE5R^:3ҋG^,# ;hç!\^ɦdBh`j4WsNV^Gqk@+UvrL_ ./lH7 ~L1oL[W2 ܇J4CU f Ee2/à~£GI/{ /t)#E)َɄ֫KXSE,x᭽+g0PK!\Җoocjk_commons/__init__.pyPK!$cjk_commons/logging_.pyPK!Z cjk_commons/settings.pyPK!#Ocjk_commons/zip.pyPK!HڽTU!| cjk_commons-3.3.3.dist-info/WHEELPK!HO"Eq$!cjk_commons-3.3.3.dist-info/METADATAPK!H%eW""cjk_commons-3.3.3.dist-info/RECORDPKY$