更新工具类

zhanghe06 · zhanghe06 · commit 4ad695a08b88 · 2017-05-18T15:39:29.000+08:00
diff --git a/test/test_except.py b/test/test_except.py
@@ -11,28 +11,41 @@
 import time
 import traceback
 
-try:
-    # raise Exception('error_message')
-    raise Exception('error_message', 'error_code')
-except Exception as e:
-    # 打印异常
-    print type(e), e
 
-    time.sleep(0.1)
+def test_01():
+    try:
+        # raise Exception('error_message')
+        raise Exception('error_message', 'error_code')
+    except Exception as e:
+        # 打印异常
+        print type(e), e
 
-    # 打印异常消息
-    print type(e.message), e.message
+        time.sleep(0.1)
 
-    time.sleep(0.1)
+        # 打印异常消息
+        print type(e.message), e.message
 
-    # 打印异常参数
-    print e.args, type(e.args[0]), e.args[0]
+        time.sleep(0.1)
 
-    time.sleep(0.1)
+        # 打印异常参数
+        print e.args, type(e.args[0]), e.args[0]
 
-    # 打印异常堆栈跟踪信息 stack traceback
-    traceback.print_exc()
+        time.sleep(0.1)
+
+        # 打印异常堆栈跟踪信息 stack traceback
+        traceback.print_exc()
+
+
+def test_02():
+    try:
+        print u'try'
+        raise Exception(u'try')
+    except Exception as e:
+        print u'except'
+        raise Exception(u'except')
+    finally:
+        print u'finally'
 
 
 if __name__ == '__main__':
-    pass
+    test_02()
diff --git a/test/test_pandas.py b/test/test_pandas.py
@@ -69,11 +69,43 @@ def test_read_sem_excel():
     return sem_map
 
 
+def test_read_area_code_excel():
+    file_path = '全球区号.xlsx'
+    df = pd.read_excel(file_path, sheetname='code')  # sheet_name=str(0)
+    print list(df.keys())
+    # print df.to_dict()
+    # sem_map = {}
+    for i in df.values:
+        code_dict = {
+            'id': i[0],
+            'name_c': i[1],
+            'area_code': i[2],
+            'phone_pre': '+00%s' % i[2],
+            'country_area': i[3],
+            'short_code': i[4],
+            'name_e': i[5]
+        }
+        print json.dumps(code_dict, indent=4, ensure_ascii=False)+','
+    # print sem_map
+
+
+def test_read_area_code_map_excel():
+    file_path = '全球区号.xlsx'
+    df = pd.read_excel(file_path, sheetname='code')  # sheet_name=str(0)
+    print list(df.keys())
+    # print df.to_dict()
+    # sem_map = {}
+    for i in df.values:
+        print '%s: \'%s\',  # [%s]%s(%s) %s' % (i[0], i[2], i[4], i[1], i[5], i[3])
+
+
 if __name__ == '__main__':
     # test_csv()
     # test_excel()
     # test_csv_encode()
-    test_read_sem_excel()
+    # test_read_sem_excel()
+    # test_read_area_code_excel()
+    test_read_area_code_map_excel()
 
 
 """
diff --git a/test/全球区号.xlsx b/test/全球区号.xlsx
diff --git a/tools/convert.py b/tools/convert.py
@@ -24,6 +24,7 @@ def hex_to_bytes(string):
 def test_unescape():
     # unicode编码后的汉字的解析
     str_xx = '&#21152;&#20837;&#21040;&#34;&#25105;&#30340;&#20070;&#30446;&#36873;&#21333;&#34;&#20013;'
+    str_xx = '\xe5\xbd\x93\xe5\x89\x8d\xe5\xb7\xb2\xe8\xbe\xbe\xe5\x88\xb0\xe6\x8a\x93\xe5\x8f\x96\xe9\x85\x8d\xe7\xbd\xae\xe7\x9a\x84\xe6\x9c\x80\xe5\xa4\xa7\xe9\xa1\xb5\xe7\xa0\x81'
     from HTMLParser import HTMLParser
     h = HTMLParser()
     print h.unescape(str_xx)
@@ -44,6 +45,7 @@ def test_unicode():
     # unicode_string.encode('target_encoding')
     print repr(u'\u6211\u6765\u4e86'.encode('utf-8')), u'\u6211\u6765\u4e86'.encode('utf-8')  # 每个汉子占3个字节
     print repr(u'\u6211\u6765\u4e86'.encode('GBK')), u'\u6211\u6765\u4e86'.encode('GBK')  # 每个汉子占2个字节
+    print repr('没有新数据，停止翻页')
 
 
 def test():
diff --git a/tools/redis_pub_sub.py b/tools/redis_pub_sub.py
@@ -47,6 +47,23 @@ def sub(self, k):
             # yield item
             yield item.get('data')
 
+    def p_sub(self, k):
+        """
+        PSub
+        订阅一个或多个符合给定模式的频道
+        每个模式以 * 作为匹配符
+        注意 psubscribe 与 subscribe 区别
+        :param k:
+        :return:
+        """
+        ps = self.__db.pubsub()
+        ch = '%s:%s' % (self.key, k)
+        ps.psubscribe(ch)
+        for item in ps.listen():
+            # {'pattern': None, 'type': 'subscribe', 'channel': 'queue:test:hh', 'data': 1L}
+            # yield item
+            yield item.get('data')
+
     def sub_not_loop(self, k):
         """
         Sub 非无限循环，取到结果即退出
@@ -62,17 +79,24 @@ def sub_not_loop(self, k):
 
 
 def test_pub():
-    q = RedisPubSub('test')
+    q = RedisPubSub('test:aa')
     q.pub('hh', '123')
 
 
 def test_sub():
-    q = RedisPubSub('test')
+    q = RedisPubSub('test:aa')
     r = q.sub('hh')
     for i in r:
         print i
 
 
+def test_p_sub():
+    q = RedisPubSub('test:*')
+    r = q.p_sub('hh')
+    for i in r:
+        print i
+
+
 def test_sub_not_loop():
     q = RedisPubSub('test')
     r = q.sub_not_loop('hh')
@@ -104,6 +128,14 @@ def run():
 ✗ python redis_pub_sub.py test_pub
 
 
+测试模式订阅
+终端一：
+✗ python redis_pub_sub.py test_p_sub
+终端二：
+✗ python redis_pub_sub.py test_pub
+
+
+
 测试非无限循环模式
 终端一：
 ✗ python redis_pub_sub.py test_sub_not_loop
diff --git a/tools/token.py b/tools/token.py
@@ -0,0 +1,38 @@
+#!/usr/bin/env python
+# encoding: utf-8
+
+"""
+@author: zhanghe
+@software: PyCharm
+@file: token.py
+@time: 2016/11/29 下午7:13
+"""
+
+import base64
+import json
+
+
+class Token(object):
+    def __init__(self):
+        self._key = '123qweasdzxc'
+
+    def sign(self, data):
+        data += '.%s' % self._key
+        print data
+        return base64.b64encode(data)
+
+    def unsign(self, data):
+        return base64.b64decode(data)
+
+
+if __name__ == '__main__':
+    token = Token()
+    # 测试字符串
+    a = token.sign('1234')
+    b = token.unsign(a)
+    print a, b
+    # 测试字典
+    test_dict_str = json.dumps({'z': 56, 'm': 12, 'p': 34}, sort_keys=True)
+    a = token.sign(test_dict_str)
+    b = token.unsign(a)
+    print a, b
