common.py 4.9 KB
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2020-02-10 23:12
# @Author  : Lemon
# @File    : common.py
# @Software: PyCharm

import datetime
import demjson
import base64
from enum import Enum
from attr import attrs, attrib


@attrs(cmp=False)
class WorkPlace():
    employer_name = attrib("")
    employer_id = attrib("")
    position_text = attrib("")
    position_id = attrib("")
    location_name = attrib("")
    location_id = attrib("")
    description = attrib("")
    current = attrib(False, type=bool)
    start_time = attrib(None, type=datetime.date)
    end_time = attrib(None, type=datetime.date)

    def to_dict(self):
        data = {
            'employer_name': self.employer_name,
            'employer_id': self.employer_id,
            'position_text': self.position_text,
            'position_id': self.position_id,
            'location_name': self.location_name,
            'location_id': self.location_id,
            'description': self.description,
            'date_start[year]': str(self.start_time.year),
            'date_start[month]': str(self.start_time.month),
            'date_start[day]': str(self.start_time.day),
            'date_end[year]': str(self.end_time.year) if self.end_time else '',
            'date_end[month]': str(self.end_time.month) if self.end_time else '',
            'date_end[day]': str(self.end_time.day) if self.end_time else '',
        }
        if self.current:
            data['date[current]'] = 'on'
        return data

    @classmethod
    def from_dict(cls, data: dict):
        self = cls()
        for k, v in data.items():
            if hasattr(self, k):
                setattr(self, k, v)

        start_time = []
        end_time = []
        for x in ['year', 'month', 'day']:
            if data.get(f'date_start[{x}]'):
                start_time.append(int(data.get(f'date_start[{x}]')))

            if data.get(f'date_end[{x}]'):
                end_time.append(int(data.get(f'date_end[{x}]')))
        if start_time:
            self.start_time = datetime.date(*start_time)

        if end_time:
            self.end_time = datetime.date(*end_time)
        else:
            self.current = True

        return self


@attrs(cmp=False)
class College():
    school_text = attrib("")
    school_id = attrib("")
    subject = attrib([], type=list, converter=lambda x: [] if x is None else x)
    school_type = attrib("college")
    description = attrib("")
    graduated = attrib(False, type=bool)  # 'graduated': 'on',
    start_time = attrib(None, type=datetime.date)
    end_time = attrib(None, type=datetime.date)

    def to_dict(self):
        data = {
            'school_text': self.school_text,
            'school_id': self.school_id,
            'date_start[year]': str(self.start_time.year),
            'date_start[month]': str(self.start_time.month),
            'date_start[day]': str(self.start_time.day),
            'date_end[year]': str(self.end_time.year) if self.end_time else '',
            'date_end[month]': str(self.end_time.month) if self.end_time else '',
            'date_end[day]': str(self.end_time.day) if self.end_time else '',
            'description': self.description,
            'school_type': self.school_type
        }
        if self.graduated:
            data['graduated'] = 'on'
        if self.subject:
            for i, sub in enumerate(self.subject):
                assert len(sub) == 2, '格式不对,要求格式(concentration_text,concentration_ids)'
                data["concentration_text[%d]" % i] = sub[0]
                data["concentration_ids[%d]" % i] = sub[1]
        return data


class TaskStatus():
    EXCEPTION = -2
    FAILED = -1
    EXECUTE = 0
    SUCCESS = 1


class Exchange(Enum):
    CLIENT = 0

    # SERVER = 1
    # NSQ = 2
    # MQTT = 3

    @classmethod
    def get(cls, code) -> Enum:
        if isinstance(code, str):
            code = int(code)

        keys = [x for x in dir(cls) if not x.startswith('__')]
        for key in keys:
            member = getattr(cls, key)
            if member.value == code:
                return member
        return None


def todict(obj, include: list = None):
    keys = dir(obj)
    res = {}
    if include:
        for key in include:
            try:
                value = getattr(obj, key)
                res[key] = value
            except:
                pass
        return res
    else:
        for key in keys:
            if not key.startswith('_'):
                value = getattr(obj, key)
                res[key] = value
        return res


def tobase64(obj):
    if isinstance(obj, (dict, list)):
        obj = demjson.encode(obj)
    if isinstance(obj, str):
        obj = obj.encode('utf-8')
    if isinstance(obj, bytes):
        bin = base64.b64encode(obj)
        return str(bin, 'utf-8')
    raise BaseException("must str,list,dict,bytes")


def frombase64(base):
    string = base64.b64decode(base).decode()
    return demjson.decode(string)