NovaBryan/EVE_Dump
Log | Files | 📦 下载最新源码 (.zip) | Refs | README

cerberus_formatted_message.py (4578B)

      1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\packages\localization\client\cerberus_formatted_message.py
      2 from datetime import datetime, timedelta
      3 from datetimeutils import datetime_to_filetime
      4 from eveProto.generated.eve_public.localization.cerberus.cerberus_pb2 import FormattedMessage as ProtoFormattedMessage
      5 from eveProto.generated.eve_public.localization.cerberus.cerberus_pb2 import Parameter as ProtoParameter
      6 from eveProto.generated.eve_public.localization.message_pb2 import Identifier as LocalizationIdentifier
      7 from google.protobuf.duration_pb2 import Duration
      8 from google.protobuf.timestamp_pb2 import Timestamp
      9 from localization import GetByMessageID
     10 from localization.common.errors import UnsupportedParameterKeyError, DuplicateParameterKeyError
     11 if False:
     12     from typing import Dict
     13 
     14 class CerberusFormattedMessage(object):
     15 
     16     def __init__(self, identifier, parameters):
     17         self._identifier = identifier
     18         self._parameters = parameters
     19 
     20     @property
     21     def identifier(self):
     22         return self._identifier
     23 
     24     @property
     25     def parameters(self):
     26         return self._parameters
     27 
     28     @property
     29     def formatted_parameters(self):
     30         result = {}
     31         for key, value in self._parameters.items():
     32             if isinstance(value, datetime):
     33                 result[key] = datetime_to_filetime(value)
     34             elif isinstance(value, timedelta):
     35                 result[key] = value.total_seconds()
     36             else:
     37                 result[key] = value
     38 
     39         return result
     40 
     41     def get_localized_text(self):
     42         return GetByMessageID(self.identifier, **self.formatted_parameters)
     43 
     44     @classmethod
     45     def from_proto(cls, payload):
     46         cerberus_parameters = {}
     47         for raw_param in payload.parameters:
     48             key = raw_param.key
     49             if key in cerberus_parameters:
     50                 raise DuplicateParameterKeyError('duplicate key found: %s' % key)
     51             value = None
     52             if raw_param.HasField('string'):
     53                 value = raw_param.string
     54             elif raw_param.HasField('integer'):
     55                 value = raw_param.integer
     56             elif raw_param.HasField('signed_integer'):
     57                 value = raw_param.signed_integer
     58             elif raw_param.HasField('double'):
     59                 value = raw_param.double
     60             elif raw_param.HasField('timestamp'):
     61                 value = datetime.fromtimestamp(raw_param.timestamp.ToSeconds())
     62             elif raw_param.HasField('duration'):
     63                 value = timedelta(seconds=raw_param.duration.ToSeconds())
     64             if value is None:
     65                 raise UnsupportedParameterKeyError('unsupported type for key %s' % key)
     66             cerberus_parameters[key] = value
     67 
     68         return CerberusFormattedMessage(identifier=payload.identifier.sequential, parameters=cerberus_parameters)
     69 
     70     @classmethod
     71     def to_proto(cls, instance):
     72         payload = ProtoFormattedMessage(identifier=LocalizationIdentifier(sequential=instance.identifier))
     73         for key, val in instance.parameters.items():
     74             proto_param = ProtoParameter(key=key)
     75             if isinstance(val, str):
     76                 proto_param.string = val.encode('utf-8')
     77             elif isinstance(val, int):
     78                 if val >= 0:
     79                     proto_param.integer = val
     80                 else:
     81                     proto_param.signed_integer = val
     82             elif isinstance(val, float):
     83                 proto_param.double = val
     84             elif isinstance(val, datetime):
     85                 timestamp_proto = Timestamp()
     86                 timestamp_proto.FromDatetime(val)
     87                 proto_param.timestamp.CopyFrom(timestamp_proto)
     88             elif isinstance(val, timedelta):
     89                 proto_param.duration.CopyFrom(Duration(seconds=int(val.total_seconds())))
     90             else:
     91                 raise UnsupportedParameterKeyError('unsupported type for key %s' % key)
     92             payload.parameters.append(proto_param)
     93 
     94         return payload
     95 
     96     def __eq__(self, other):
     97         if other is None:
     98             return False
     99         return self.identifier == other.identifier and self.parameters == other.parameters
    100 
    101     def __ne__(self, other):
    102         return not self.__eq__(other)
    103 
    104     def __str__(self):
    105         params_str = 'params: \n'
    106         for k, v in self.parameters.items():
    107             params_str += 'Key: %s, Value: %s\n' % (k, v)
    108 
    109         return ('id: %s, ' % self.identifier + params_str).encode('utf-8')
    110 
    111     def __repr__(self):
    112         return '<%s %r>' % (self.__class__.__name__, self.__dict__)