Исходный код jim_mes.convert

# -*- coding: utf-8 -*-
# @Author: Max ST
# @Date:   2019-04-05 01:21:21
# @Last Modified by:   MaxST
# @Last Modified time: 2019-08-08 21:43:31
import csv
import json
import logging
import tempfile

from io import StringIO
from pathlib import Path
from yaml import dump, load

try:
    from yaml import CLoader as Loader
except ImportError:
    from yaml import Loader


[документация]class PrototypeDispatcher(object): """Диспетчер прототипов форматов.""" def __init__(self): """Инициализация.""" self._objects = {}
[документация] def get_objects(self): """Get all objects.""" return self._objects
[документация] def register_object(self, name, obj): """Register an object.""" self._objects[name] = obj
[документация] def unregister_object(self, name): """Unregister an object.""" del self._objects[name]
dispatcher = PrototypeDispatcher()
[документация]class Converter(object): """Основной класс конвертера.""" def __init__(self, *args, **kwargs): """Инициализация. Args: *args: доп. параметры **kwargs: тип файла и логер """ super().__init__() self.type = kwargs.get('type', 'yaml') self.file_name = kwargs.get('file_name', None) self.logger = kwargs.get('logger', logging.getLogger(type(self).__name__))
[документация] def read(self, data=None): """Чтение. Args: data: данные (default: {None}) Returns: Прочтенные данные object """ self.file_name = data if data else self.file_name self.define_convert() self.logd(f'read from {self.type}') return self.convert().read(self.file_name)
[документация] def reads(self, data): """Чтение из строки. Args: data: данные для чтения Returns: Прочтенные данные object """ self.define_convert() self.logd(f'read from {self.type}') return self.convert().reads(data)
[документация] def write(self, data, type_=None): """Запись данных. Args: data: Данные type_: тип (default: {None}) Returns: [description] [type] """ if type_: self.type = type_ self.file_name = None self.define_convert() self.logd(f'write to {self.type}') self.file_name = self.convert().write(data) self.logd(self.file_name) return self.file_name
[документация] def dumps(self, data): """Сброс в строку.""" self.define_convert() self.logd(f'dumps to {self.type}') return self.convert().dumps(data)
[документация] def define_convert(self): """Определение конвертера.""" if self.file_name: path = Path(self.file_name) self.type = path.suffix.strip('.') if self.type: self.convert = dispatcher.get_objects().get(self.type)
[документация] def logd(self, *args): """Логирование.""" self.logger.debug(*args)
[документация] def convert_file_to(self, file_name, to_type='csv'): """Конвертирование в файл. Args: file_name: имя файла to_type: целевой тип (default: {'csv'}) Returns: имя файла str """ self.file_name = file_name self.write(self.read(), to_type) return self.file_name
[документация]class Csv(object): """Конвертер CSV."""
[документация] def read(self, file_name): """Чтение из файла. Args: file_name: Имя файла """ with Path(file_name).open() as f: reader = csv.DictReader(f) return [r for r in reader]
[документация] def write(self, data): """Запись во временный файл. Args: data: Данные для записи Returns: Имя файла str """ response = None writer = None with tempfile.NamedTemporaryFile(mode='w', prefix='test_file', suffix='.csv', delete=False) as ntf: for d in data: if not writer: writer = csv.DictWriter(ntf, fieldnames=d.keys()) writer.writeheader() break writer.writerows(data) response = ntf.name return response
[документация] def reads(self, data): """Чтение из строки. Args: data: данные для чтения """ f = StringIO(data) reader = csv.DictReader(f) return [r for r in reader]
[документация] def dumps(self, data): """Данные в строку. Args: data: Данные для преобразования Returns: [description] [type] """ stream = StringIO() getvalue = stream.getvalue writer = None for d in data: if not writer: writer = csv.DictWriter(stream, fieldnames=d.keys()) writer.writeheader() break writer.writerows(data) return getvalue()
[документация]class Json(object): """Конвертер JSON."""
[документация] def read(self, file_name): with Path(file_name).open() as f: return json.load(f)
[документация] def reads(self, data): return json.loads(data)
[документация] def write(self, data): response = None with tempfile.NamedTemporaryFile(mode='w', prefix='test_file', suffix='.json', delete=False) as ntf: response = ntf.name json.dump(data, ntf, sort_keys=True, indent=4, ensure_ascii=False) return response
[документация] def dumps(self, data): return json.dumps(data)
[документация]class Yaml(object): """Конвертер YAML."""
[документация] def read(self, file_name): with Path(file_name).open() as f: return load(f, Loader=Loader)
[документация] def reads(self, data): f = StringIO(data) return load(f, Loader=Loader)
[документация] def write(self, data): response = None with tempfile.NamedTemporaryFile(mode='w', prefix='test_file', suffix='.yaml', delete=False) as ntf: response = ntf.name dump(data, ntf, default_flow_style=False, allow_unicode=True, indent=4) return response
[документация] def dumps(self, data): return dump(data, default_flow_style=False, allow_unicode=True, indent=4)
dispatcher.register_object('csv', Csv) dispatcher.register_object('json', Json) dispatcher.register_object('yaml', Yaml)