如何查找流量回执消息（2）

编写样例代码

1. #!/usr/bin/env python
2. # coding=utf8
3. import sys
4. from aliyunsdkcore.acs_exception.exceptions import ServerException
5. from aliyunsdkcore.client import AcsClient
6. from aliyunsdkdybaseapi.request.v20170525.QueryTokenForMnsQueueRequest import QueryTokenForMnsQueueRequest
8. import datetime
9. from mns.account import Account
10. from mns.queue import *
12. try:
13.     import json
14. except ImportError:
15.     import simplejson as json
17. """
18. 云通信基础能力业务回执消息消费示例，供参考。
19. Created on 2017-06-13
20. """
23. reload(sys)
24. sys.setdefaultencoding('utf8')
25. #暂不支持多region,默认需配置成cn-hangzhou
26. regionid = "cn-hangzhou"
28. # ACCESS_KEY_ID/ACCESS_KEY_SECRET 根据实际申请的账号信息进行替换
29. accesskeyid = "yourAccessKeyId"
30. accesskeysecret = "yourAccessKeySecret"
31. endpoint = "http://1943695596114318.mns.cn-hangzhou.aliyuncs.com"
33. # 短信回执：SmsReport，短息上行：SmsUp，语音呼叫：VoiceReport，流量直冲：FlowReport
34. msgtype = "FlowReport"
35. # 在云通信页面开通相应业务消息后，就能在页面上获得对应的queueName
36. qname = "Alicom-Queue-1093695199815616-FlowReport"
38. acs_client = AcsClient(accesskeyid, accesskeysecret, regionid)
41. # 云通信业务token存在失效时间，需动态更新。
42. class Token():
43.     def __init__(self, token=None, tmp_access_id=None, tmp_access_key=None, expire_time=None):
44.         self.__token = token
45.         self.__tmp_access_id = tmp_access_id
46.         self.__tmp_access_key = tmp_access_key
47.         self.__expire_time = expire_time
49.     def get_token(self):
50.         return self.__token
52.     def set_token(self, token):
53.         self.__token = token
55.     def get_tmp_access_id(self):
56.         return self.__tmp_access_id
58.     def set_tmp_access_id(self, tmp_access_id):
59.         self.__tmp_access_id = tmp_access_id
61.     def get_tmp_access_key(self):
62.         return self.__tmp_access_key
64.     def set_tmp_access_key(self, tmp_access_key):
65.         self.__tmp_access_key = tmp_access_key
67.     def get_expire_time(self):
68.         return self.__expire_time
70.     def set_expire_time(self, expire_time):
71.         self.__expire_time = expire_time
73.     def is_refresh(self):
74.         # 失效时间与当前系统时间比较，提前2分钟刷新token
75.         now = datetime.datetime.now()
76.         expire = datetime.datetime.strptime(self.__expire_time, "%Y-%m-%d %H:%M:%S")
77.         # intval = (expire - now).seconds
78.         # print "token生效剩余时长（秒）：" + str(intval)
79.         if (expire - now).seconds < 120:
80.             return 1
81.         return 0
83.     def refresh(self):
84.         print "start refresh token..."
85.         request = QueryTokenForMnsQueueRequest()
86.         request.set_MessageType(msgtype)
87.         response = acs_client.do_action_with_exception(request)
88.         # print response
89.         if response is None:
90.             raise ServerException("GET_TOKEN_FAIL", "获取token时无响应")
92.         response_body = json.loads(response)
94.         if response_body.get("Code") != "OK":
95.             raise ServerException("GET_TOKEN_FAIL", "获取token失败")
97.         self.__tmp_access_key = response_body.get("MessageTokenDTO").get("AccessKeySecret")
98.         self.__tmp_access_id = response_body.get("MessageTokenDTO").get("AccessKeyId")
99.         self.__expire_time = response_body.get("MessageTokenDTO").get("ExpireTime")
100.         self.__token = response_body.get("MessageTokenDTO").get("SecurityToken")
102.         print "finsh refresh token..."
105. # 初始化 my_account, my_queue
106. token = Token()
107. token.refresh()
108. my_account = Account(endpoint, token.get_tmp_access_id(), token.get_tmp_access_key(), token.get_token())
109. my_queue = my_account.get_queue(qname)
110. # my_queue.set_encoding(False)
111. # 循环读取删除消息直到队列空
112. # receive message请求使用long polling方式，通过wait_seconds指定长轮询时间为3秒
114. ## long polling 解析:
115. ### 当队列中有消息时，请求立即返回；
116. ### 当队列中没有消息时，请求在MNS服务器端挂3秒钟，在这期间，有消息写入队列，请求会立即返回消息，3秒后，请求返回队列没有消息；
118. wait_seconds = 3
119. print "%sReceive And Delete Message From Queue%s\nQueueName:%s\nWaitSeconds:%s\n" % (
120.     10 * "=", 10 * "=", qname, wait_seconds)
121. while True:
122.     # 读取消息
123.     try:
124.         if token.is_refresh() == 1:
125.             # 刷新token
126.             token.refresh()
127.             my_account.mns_client.close_connection();
128.             my_account = Account(endpoint, token.get_tmp_access_id(), token.get_tmp_access_key(), token.get_token())
129.             my_queue = my_account.get_queue(qname)
130.             # my_queue.set_encoding(False)
131.         recv_msg = my_queue.receive_message(wait_seconds)
132.         print "Receive Message Succeed! ReceiptHandle:%s MessageBody:%s MessageID:%s" % (
133.             recv_msg.receipt_handle, recv_msg.message_body, recv_msg.message_id)
134.     except MNSExceptionBase, e:
135.         if e.type == "QueueNotExist":
136.             print "Queue not exist, please create queue before receive message."
137.             sys.exit(0)
138.         elif e.type == "MessageNotExist":
139.             print "Queue is empty! sleep 10s"
140.             time.sleep(10)
141.             continue
142.         print "Receive Message Fail! Exception:%s\n" % e
143.         continue
145.     # 删除消息
146.     try:
147.         my_queue.delete_message(recv_msg.receipt_handle)
148.         print "Delete Message Succeed!  ReceiptHandle:%s" % recv_msg.receipt_handle
149.     except MNSExceptionBase, e:
150.         print "Delete Message Fail! Exception:%s\n" % e