wechatpy.replies.create_reply() - Code Metrics - Inspection of "Migrate CI to GitHub Actions (#528)" - jxtech/wechatpy - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( ff4459...addf59 )

by Messense

created 2020-03-06 10:25 UTC

wechatpy.replies.create_reply() C

↳ Parent: wechatpy.replies

Complexity

Conditions

Size

Total Lines	28
Code Lines	22

Duplication

Lines	28
Ratio	100 %

Code Coverage

Tests	16
CRAP Score	9.111

Importance

Changes

Metric	Value
cc	9
eloc	22
nop	3
dl	28
loc	28
rs	6.6666
c	0
b	0
f	0
ccs	16
cts	18
cp	0.8889
crap	9.111

# -*- coding: utf-8 -*-
"""
    wechatpy.replies
    ~~~~~~~~~~~~~~~~~~
    This module defines all kinds of replies you can send to WeChat

    :copyright: (c) 2014 by messense.
    :license: MIT, see LICENSE for more details.
"""
from __future__ import absolute_import, unicode_literals
import time
import six
import xmltodict

from wechatpy.fields import (
    StringField,
    IntegerField,
    ImageField,
    VoiceField,
    VideoField,
    MusicField,
    ArticlesField,
    Base64EncodeField,
    HardwareField,
)
from wechatpy.messages import BaseMessage, MessageMetaClass
from wechatpy.utils import to_text, to_binary


REPLY_TYPES = {}


def register_reply(reply_type):
    def register(cls):
        REPLY_TYPES[reply_type] = cls
        return cls
    return register


class BaseReply(six.with_metaclass(MessageMetaClass)):
    """Base class for all replies"""
    source = StringField('FromUserName')
    target = StringField('ToUserName')
    time = IntegerField('CreateTime', time.time())

    type = 'unknown'

    def __init__(self, **kwargs):
        self._data = {}
        message = kwargs.pop('message', None)
        if message and isinstance(message, BaseMessage):
            if 'source' not in kwargs:
                kwargs['source'] = message.target
            if 'target' not in kwargs:
                kwargs['target'] = message.source
            if hasattr(message, 'agent') and 'agent' not in kwargs:
                kwargs['agent'] = message.agent
        if 'time' not in kwargs:
            kwargs['time'] = time.time()
        for name, value in kwargs.items():
            field = self._fields.get(name)
            if field:
                self._data[field.name] = value
            else:
                setattr(self, name, value)

    def render(self):
        """Render reply from Python object to XML string"""
        tpl = '<xml>\n{data}\n</xml>'
        nodes = []
        msg_type = '<MsgType><![CDATA[{msg_type}]]></MsgType>'.format(
            msg_type=self.type
        )
        nodes.append(msg_type)
        for name, field in self._fields.items():
            value = getattr(self, name, field.default)
            node_xml = field.to_xml(value)
            nodes.append(node_xml)
        data = '\n'.join(nodes)
        return tpl.format(data=data)

    def __str__(self):
        if six.PY2:
            return to_binary(self.render())
        else:
            return to_text(self.render())


@register_reply('empty')
class EmptyReply(BaseReply):
    """
    回复空串

    微信服务器不会对此作任何处理，并且不会发起重试
    """
    def __init__(self):
        pass

    def render(self):
        return ''


@register_reply('text')
class TextReply(BaseReply):
    """
    文本回复
    详情请参阅 http://mp.weixin.qq.com/wiki/9/2c15b20a16019ae613d413e30cac8ea1.html
    """
    type = 'text'
    content = StringField('Content')


@register_reply('image')
class ImageReply(BaseReply):
    """
    图片回复
    详情请参阅
    http://mp.weixin.qq.com/wiki/9/2c15b20a16019ae613d413e30cac8ea1.html
    """
    type = 'image'
    image = ImageField('Image')

    @property
    def media_id(self):
        return self.image

    @media_id.setter
    def media_id(self, value):
        self.image = value


@register_reply('voice')
class VoiceReply(BaseReply):
    """
    语音回复
    详情请参阅
    http://mp.weixin.qq.com/wiki/9/2c15b20a16019ae613d413e30cac8ea1.html
    """
    type = 'voice'
    voice = VoiceField('Voice')

    @property
    def media_id(self):
        return self.voice

    @media_id.setter
    def media_id(self, value):
        self.voice = value


@register_reply('video')
class VideoReply(BaseReply):
    """
    视频回复
    详情请参阅
    http://mp.weixin.qq.com/wiki/9/2c15b20a16019ae613d413e30cac8ea1.html
    """
    type = 'video'
    video = VideoField('Video', {})

    @property
    def media_id(self):
        return self.video.get('media_id')

    @media_id.setter
    def media_id(self, value):
        video = self.video
        video['media_id'] = value
        self.video = video

    @property
    def title(self):
        return self.video.get('title')

    @title.setter
    def title(self, value):
        video = self.video
        video['title'] = value
        self.video = video

    @property
    def description(self):
        return self.video.get('description')

    @description.setter
    def description(self, value):
        video = self.video
        video['description'] = value
        self.video = video


@register_reply('music')
class MusicReply(BaseReply):
    """
    音乐回复
    详情请参阅
    http://mp.weixin.qq.com/wiki/9/2c15b20a16019ae613d413e30cac8ea1.html
    """
    type = 'music'
    music = MusicField('Music', {})

    @property
    def thumb_media_id(self):
        return self.music.get('thumb_media_id')

    @thumb_media_id.setter
    def thumb_media_id(self, value):
        music = self.music
        music['thumb_media_id'] = value
        self.music = music

    @property
    def title(self):
        return self.music.get('title')

    @title.setter
    def title(self, value):
        music = self.music
        music['title'] = value
        self.music = music

    @property
    def description(self):
        return self.music.get('description')

    @description.setter
    def description(self, value):
        music = self.music
        music['description'] = value
        self.music = music

    @property
    def music_url(self):
        return self.music.get('music_url')

    @music_url.setter
    def music_url(self, value):
        music = self.music
        music['music_url'] = value
        self.music = music

    @property
    def hq_music_url(self):
        return self.music.get('hq_music_url')

    @hq_music_url.setter
    def hq_music_url(self, value):
        music = self.music
        music['hq_music_url'] = value
        self.music = music


@register_reply('news')
class ArticlesReply(BaseReply):
    """
    图文回复
    详情请参阅
    http://mp.weixin.qq.com/wiki/9/2c15b20a16019ae613d413e30cac8ea1.html
    """
    type = 'news'
    articles = ArticlesField('Articles', [])

    def add_article(self, article):
        if len(self.articles) == 10:
            raise AttributeError("Can't add more than 10 articles"
                                 " in an ArticlesReply")
        articles = self.articles
        articles.append(article)
        self.articles = articles


@register_reply('transfer_customer_service')
class TransferCustomerServiceReply(BaseReply):
    """
    将消息转发到多客服
    详情请参阅
    http://mp.weixin.qq.com/wiki/5/ae230189c9bd07a6b221f48619aeef35.html
    """
    type = 'transfer_customer_service'


@register_reply('device_text')
class DeviceTextReply(BaseReply):
    type = 'device_text'
    device_type = StringField('DeviceType')
    device_id = StringField('DeviceID')
    session_id = StringField('SessionID')
    content = Base64EncodeField('Content')


@register_reply('device_event')
class DeviceEventReply(BaseReply):
    type = 'device_event'
    event = StringField('Event')
    device_type = StringField('DeviceType')
    device_id = StringField('DeviceID')
    session_id = StringField('SessionID')
    content = Base64EncodeField('Content')


@register_reply('device_status')
class DeviceStatusReply(BaseReply):
    type = 'device_status'
    device_type = StringField('DeviceType')
    device_id = StringField('DeviceID')
    status = IntegerField('DeviceStatus')


@register_reply('hardware')
class HardwareReply(BaseReply):
    type = 'hardware'
    func_flag = IntegerField('FuncFlag', 0)
    hardware = HardwareField('HardWare')


def create_reply(reply, message=None, render=False):

    """
    Create a reply quickly
    """
    r = None
    if not reply:
        r = EmptyReply()
    elif isinstance(reply, BaseReply):
        r = reply
        if message:
            r.source = message.target
            r.target = message.source
    elif isinstance(reply, six.string_types):
        r = TextReply(
            message=message,
            content=reply
        )
    elif isinstance(reply, (tuple, list)):
        if len(reply) > 10:
            raise AttributeError("Can't add more than 10 articles"
                                 " in an ArticlesReply")
        r = ArticlesReply(
            message=message,
            articles=reply
        )
    if r and render:
        return r.render()
    return r


def deserialize_reply(xml, update_time=False):
    """
    反序列化被动回复
    :param xml: 待反序列化的xml
    :param update_time: 是否用当前时间替换xml中的时间
    :raises ValueError: 不能辨识的reply xml
    :rtype: wechatpy.replies.BaseReply
    """
    if not xml:
        return EmptyReply()

    try:
        reply_dict = xmltodict.parse(xml)["xml"]
        msg_type = reply_dict["MsgType"]
    except (xmltodict.expat.ExpatError, KeyError):
        raise ValueError("bad reply xml")
    if msg_type not in REPLY_TYPES:
        raise ValueError("unknown reply type")

    cls = REPLY_TYPES[msg_type]
    kwargs = dict()
    for attr, field in cls._fields.items():
        if field.name in reply_dict:
            str_value = reply_dict[field.name]
            kwargs[attr] = field.from_xml(str_value)

    if update_time:
        kwargs["time"] = time.time()

    return cls(**kwargs)


1			# -- coding: utf-8 --
2	10		"""
3			wechatpy.replies
4			~~~~~~~~~~~~~~~~~~
5			This module defines all kinds of replies you can send to WeChat
6
7			:copyright: (c) 2014 by messense.
8			:license: MIT, see LICENSE for more details.
9			"""
10	10		from __future__ import absolute_import, unicode_literals
11	10		import time
12	10		import six
13	10		import xmltodict
14
15	10		from wechatpy.fields import (
16			StringField,
17			IntegerField,
18			ImageField,
19			VoiceField,
20			VideoField,
21			MusicField,
22			ArticlesField,
23			Base64EncodeField,
24			HardwareField,
25			)
26	10		from wechatpy.messages import BaseMessage, MessageMetaClass
27	10		from wechatpy.utils import to_text, to_binary
28
29
30	10		REPLY_TYPES = {}
31
32
33	10		def register_reply(reply_type):
34	10		def register(cls):
35	10		REPLY_TYPES[reply_type] = cls
36	10		return cls
37	10		return register
38
39
40	10		class BaseReply(six.with_metaclass(MessageMetaClass)):
41			"""Base class for all replies"""
42	10		source = StringField('FromUserName')
43	10		target = StringField('ToUserName')
44	10		time = IntegerField('CreateTime', time.time())
			0 ignored issues – show Comprehensibility Best Practice introduced 2020-03-06 10:29 UTC by Report Bug Copy Issue Report The variable `time` does not seem to be defined. Loading history...
45	10		type = 'unknown'
46
47	10		def __init__(self, **kwargs):
48	10		self._data = {}
49	10		message = kwargs.pop('message', None)
50	10		if message and isinstance(message, BaseMessage):
51	10		if 'source' not in kwargs:
52	10		kwargs['source'] = message.target
53	10		if 'target' not in kwargs:
54	10		kwargs['target'] = message.source
55	10		if hasattr(message, 'agent') and 'agent' not in kwargs:
56	10		kwargs['agent'] = message.agent
57	10		if 'time' not in kwargs:
58	10		kwargs['time'] = time.time()
59	10		for name, value in kwargs.items():
60	10		field = self._fields.get(name)
61	10		if field:
62	10		self._data[field.name] = value
63			else:
64	10		setattr(self, name, value)
65
66	10		def render(self):
67			"""Render reply from Python object to XML string"""
68	10		tpl = '<xml>\n{data}\n</xml>'
69	10		nodes = []
70	10		msg_type = '<MsgType><![CDATA[{msg_type}]]></MsgType>'.format(
71			msg_type=self.type
72			)
73	10		nodes.append(msg_type)
74	10		for name, field in self._fields.items():
75	10		value = getattr(self, name, field.default)
76	10		node_xml = field.to_xml(value)
77	10		nodes.append(node_xml)
78	10		data = '\n'.join(nodes)
79	10		return tpl.format(data=data)
80
81	10		def __str__(self):
82			if six.PY2:
83			return to_binary(self.render())
84			else:
85			return to_text(self.render())
86
87
88	10		@register_reply('empty')
89	10		class EmptyReply(BaseReply):
90			"""
91			回复空串
92
93			微信服务器不会对此作任何处理，并且不会发起重试
94			"""
95	10		def __init__(self):
96	10		pass
97
98	10		def render(self):
99	10		return ''
100
101
102	10		@register_reply('text')
103	10		class TextReply(BaseReply):
104			"""
105			文本回复
106			详情请参阅 http://mp.weixin.qq.com/wiki/9/2c15b20a16019ae613d413e30cac8ea1.html
107			"""
108	10		type = 'text'
109	10		content = StringField('Content')
110
111
112	10		@register_reply('image')
113	10		class ImageReply(BaseReply):
114			"""
115			图片回复
116			详情请参阅
117			http://mp.weixin.qq.com/wiki/9/2c15b20a16019ae613d413e30cac8ea1.html
118			"""
119	10		type = 'image'
120	10		image = ImageField('Image')
121
122	10		@property
123			def media_id(self):
124	10		return self.image
125
126	10		@media_id.setter
127			def media_id(self, value):
128	10		self.image = value
129
130
131	10		@register_reply('voice')
132	10		class VoiceReply(BaseReply):
133			"""
134			语音回复
135			详情请参阅
136			http://mp.weixin.qq.com/wiki/9/2c15b20a16019ae613d413e30cac8ea1.html
137			"""
138	10		type = 'voice'
139	10		voice = VoiceField('Voice')
140
141	10		@property
142			def media_id(self):
143	10		return self.voice
144
145	10		@media_id.setter
146			def media_id(self, value):
147	10		self.voice = value
148
149
150	10		@register_reply('video')
151	10		class VideoReply(BaseReply):
152			"""
153			视频回复
154			详情请参阅
155			http://mp.weixin.qq.com/wiki/9/2c15b20a16019ae613d413e30cac8ea1.html
156			"""
157	10		type = 'video'
158	10		video = VideoField('Video', {})
159
160	10		@property
161			def media_id(self):
162	10		return self.video.get('media_id')
163
164	10		@media_id.setter
165			def media_id(self, value):
166	10		video = self.video
167	10		video['media_id'] = value
168	10		self.video = video
169
170	10		@property
171			def title(self):
172	10		return self.video.get('title')
173
174	10		@title.setter
175			def title(self, value):
176	10		video = self.video
177	10		video['title'] = value
178	10		self.video = video
179
180	10		@property
181			def description(self):
182			return self.video.get('description')
183
184	10		@description.setter
185			def description(self, value):
186	10		video = self.video
187	10		video['description'] = value
188	10		self.video = video
189
190
191	10		@register_reply('music')
192	10		class MusicReply(BaseReply):
193			"""
194			音乐回复
195			详情请参阅
196			http://mp.weixin.qq.com/wiki/9/2c15b20a16019ae613d413e30cac8ea1.html
197			"""
198	10		type = 'music'
199	10		music = MusicField('Music', {})
200
201	10		@property
202			def thumb_media_id(self):
203	10		return self.music.get('thumb_media_id')
204
205	10		@thumb_media_id.setter
206			def thumb_media_id(self, value):
207	10		music = self.music
208	10		music['thumb_media_id'] = value
209	10		self.music = music
210
211	10		@property
212			def title(self):
213	10		return self.music.get('title')
214
215	10		@title.setter
216			def title(self, value):
217	10		music = self.music
218	10		music['title'] = value
219	10		self.music = music
220
221	10		@property
222			def description(self):
223	10		return self.music.get('description')
224
225	10		@description.setter
226			def description(self, value):
227	10		music = self.music
228	10		music['description'] = value
229	10		self.music = music
230
231	10		@property
232			def music_url(self):
233	10		return self.music.get('music_url')
234
235	10		@music_url.setter
236			def music_url(self, value):
237	10		music = self.music
238	10		music['music_url'] = value
239	10		self.music = music
240
241	10		@property
242			def hq_music_url(self):
243	10		return self.music.get('hq_music_url')
244
245	10		@hq_music_url.setter
246			def hq_music_url(self, value):
247	10		music = self.music
248	10		music['hq_music_url'] = value
249	10		self.music = music
250
251
252	10		@register_reply('news')
253	10		class ArticlesReply(BaseReply):
254			"""
255			图文回复
256			详情请参阅
257			http://mp.weixin.qq.com/wiki/9/2c15b20a16019ae613d413e30cac8ea1.html
258			"""
259	10		type = 'news'
260	10		articles = ArticlesField('Articles', [])
261
262	10		def add_article(self, article):
263	10		if len(self.articles) == 10:
264			raise AttributeError("Can't add more than 10 articles"
265			" in an ArticlesReply")
266	10		articles = self.articles
267	10		articles.append(article)
268	10		self.articles = articles
269
270
271	10		@register_reply('transfer_customer_service')
272	10		class TransferCustomerServiceReply(BaseReply):
273			"""
274			将消息转发到多客服
275			详情请参阅
276			http://mp.weixin.qq.com/wiki/5/ae230189c9bd07a6b221f48619aeef35.html
277			"""
278	10		type = 'transfer_customer_service'
279
280
281	10		@register_reply('device_text')
282	10		class DeviceTextReply(BaseReply):
283	10		type = 'device_text'
284	10		device_type = StringField('DeviceType')
285	10		device_id = StringField('DeviceID')
286	10		session_id = StringField('SessionID')
287	10		content = Base64EncodeField('Content')
288
289
290	10		@register_reply('device_event')
291	10		class DeviceEventReply(BaseReply):
292	10		type = 'device_event'
293	10		event = StringField('Event')
294	10		device_type = StringField('DeviceType')
295	10		device_id = StringField('DeviceID')
296	10		session_id = StringField('SessionID')
297	10		content = Base64EncodeField('Content')
298
299
300	10		@register_reply('device_status')
301	10		class DeviceStatusReply(BaseReply):
302	10		type = 'device_status'
303	10		device_type = StringField('DeviceType')
304	10		device_id = StringField('DeviceID')
305	10		status = IntegerField('DeviceStatus')
306
307
308	10		@register_reply('hardware')
309	10		class HardwareReply(BaseReply):
310	10		type = 'hardware'
311	10		func_flag = IntegerField('FuncFlag', 0)
312	10		hardware = HardwareField('HardWare')
313
314
315	10	View Code Duplication	def create_reply(reply, message=None, render=False):
			0 ignored issues – show Duplication introduced 2020-03-06 10:29 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
316			"""
317			Create a reply quickly
318			"""
319	10		r = None
320	10		if not reply:
321	10		r = EmptyReply()
322	10		elif isinstance(reply, BaseReply):
323	10		r = reply
324	10		if message:
325			r.source = message.target
326			r.target = message.source
327	10		elif isinstance(reply, six.string_types):
328	10		r = TextReply(
329			message=message,
330			content=reply
331			)
332	10		elif isinstance(reply, (tuple, list)):
333	10		if len(reply) > 10:
334	10		raise AttributeError("Can't add more than 10 articles"
335			" in an ArticlesReply")
336	10		r = ArticlesReply(
337			message=message,
338			articles=reply
339			)
340	10		if r and render:
341	10		return r.render()
342	10		return r
343
344
345	10		def deserialize_reply(xml, update_time=False):
346			"""
347			反序列化被动回复
348			:param xml: 待反序列化的xml
349			:param update_time: 是否用当前时间替换xml中的时间
350			:raises ValueError: 不能辨识的reply xml
351			:rtype: wechatpy.replies.BaseReply
352			"""
353	10		if not xml:
354	10		return EmptyReply()
355
356	10		try:
357	10		reply_dict = xmltodict.parse(xml)["xml"]
358	10		msg_type = reply_dict["MsgType"]
359			except (xmltodict.expat.ExpatError, KeyError):
360			raise ValueError("bad reply xml")
361	10		if msg_type not in REPLY_TYPES:
362			raise ValueError("unknown reply type")
363
364	10		cls = REPLY_TYPES[msg_type]
365	10		kwargs = dict()
366	10		for attr, field in cls._fields.items():
367	10		if field.name in reply_dict:
368	10		str_value = reply_dict[field.name]
369	10		kwargs[attr] = field.from_xml(str_value)
370
371	10		if update_time:
372			kwargs["time"] = time.time()
373
374			return cls(**kwargs)
375

jxtech / wechatpy

Push — master ( ff4459...addf59 )

wechatpy.replies.create_reply() C

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like