Merge 0d9abf6 into deec830

flask_weixin.py

@@ -27,6 +27,7 @@
 
 
 class Weixin(object):
+
     """Interface for mp.weixin.qq.com
 
     http://mp.weixin.qq.com/wiki/index.php
@@ -199,6 +200,25 @@ def reply(self, username, type='text', sender=None, **kwargs):
 
         return None
 
+    def transfer_customer_service(self, username, sender=None, to_kf=None):
+        """Transfer messages to customer service.
+
+           You can transfer messages to customer service, or assign it to a
+           specific customer service.
+
+           * username: the receiver's username
+           * sender: sender is optional if you have a default value
+           * to_kf: the customer service account. Messages will assigned to
+                    specific customer service if it not None.
+        """
+        if not sender:
+            sender = self.sender
+
+        if not sender:
+            raise RuntimeError('WEIXIN_SENDER is missing')
+
+        return transfer_customer_service_reply(username, sender, to_kf)
+
     def register(self, key=None, func=None, **kwargs):
         """Register a command helper function.
 
@@ -361,6 +381,25 @@ def news_reply(username, sender, *items):
     return template % dct
 
 
+def transfer_customer_service_reply(username, sender, to_kf):
+    template = (
+        '<xml>%(shared)s'
+        '%(transfer_info)s</xml>')
+    transfer_info = ''
+    if to_kf:
+        transfer_info = (
+            '<TransInfo>'
+            '<KfAccount>![CDATA[%s]]</KfAccount>'
+            '</TransInfo>') % to_kf
+
+    dct = {
+        'shared': _shared_reply(username, sender,
+                                type='transfer_customer_service'),
+        'transfer_info': transfer_info
+    }
+    return template % dct
+
+
 def _shared_reply(username, sender, type):
     dct = {
         'username': username,

test_weixin.py

@@ -271,6 +271,59 @@ def print_send(*args, **kwargs):
         self.client.post(signature_url, data=text)
 
 
+class TestTransferCustomerServiceWeixin(Base):
+    '''
+    <xml>
+    <ToUserName><![CDATA[toUser]]></ToUserName>
+    <FromUserName><![CDATA[fromUser]]></FromUserName>
+    <CreateTime>1348831860</CreateTime>
+    <MsgType><![CDATA[text]]></MsgType>
+    <Content><![CDATA[%s]]></Content>
+    <MsgId>1234567890123456</MsgId>
+    </xml>
+    '''
+
+    def setup_weixin(self):
+        weixin = self.weixin
+
+        def print_all(**kwargs):
+            username = kwargs.get('sender')
+            sender = kwargs.get('receiver')
+            content = kwargs.get('content')
+            if not content:
+                content = 'to_kf'
+            if content == 'to_kf':
+                return weixin.transfer_customer_service(
+                    username, sender=sender)
+            if content == 'to_foo':
+                return weixin.transfer_customer_service(
+                    username, sender=sender, to_kf='foo@bar')
+
+        weixin.register('*', print_all)
+
+    def test_transfer_customer_services(self):
+        text = self.__doc__ % 'to_kf'
+        rv = self.client.post(signature_url, data=text)
+        assert b'transfer_customer_service' in rv.data
+
+    def test_transfer_to_a_customer_service(self):
+        text = self.__doc__ % 'to_foo'
+        rv = self.client.post(signature_url, data=text)
+        assert b'transfer_customer_service' in rv.data
+        assert b'foo@bar' in rv.data
+
+    @raises(RuntimeError)
+    def test_no_sender(self):
+        @self.weixin.register('send')
+        def print_send(*args, **kwargs):
+            username = kwargs.get('sender')
+            return self.weixin.transfer_customer_service(
+                username, sender=None)
+
+        text = self.__doc__ % 'send'
+        self.client.post(signature_url, data=text)
+
+
 class TestKeyMatching(Base):
 
     def setup_weixin(self):