分享有礼：Python代码的魔鬼细节大比拼！（获奖名单已公布-2014-8-11）

shenlanyouyu

对Python有一点了解，LP在工作中用得比较多，支持活动，顶一个。

wsysx

好活动  必须顶啊，对python只是偶尔能简单用点

txdgtwpv

前几天写的HTTP代理服务器  lightproxy.py, 除掉空行在100行以内

1. 
   
2. #!/usr/bin/env python
   
3. from select import select
   
4. from socket import socket
   
5. from sys import argv, exit
   
6. from urlparse import urlparse, urlunparse
   
7. from BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer
   
8. from SocketServer import ThreadingMixIn
   
9. 
   
10. __version__ = '0.01'
    
11. bufsize = 1024 * 1024  # 1MB
    
12. timeout = 60
    
13. 
    
14. 
    
15. class ProxyHandler(BaseHTTPRequestHandler):
    
16.     server_version = 'SimpleProxy/%s' % __version__
    
17. 
    
18.     def do_CONNECT(self):
    
19.         self.log_request()
    
20.         sock = socket()
    
21.         self._connect_remote(sock, self.path)
    
22.         self.wfile.write('%s 200 Connection Established\r\n'
    
23.                          % self.protocol_version)
    
24.         self.wfile.write('Proxy-agent: %s\r\n' % self.version_string())
    
25.         self.wfile.write('\r\n')
    
26.         self._do_proxy(sock)
    
27.         self.connection.close()
    
28.         sock.close()
    
29. 
    
30.     def do_GET(self):
    
31.         import time
    
32.         print time.ctime()
    
33.         self.log_request()
    
34.         res = urlparse(self.path)
    
35.         if res.scheme != 'http' or res.fragment or not res.netloc:
    
36.             self.send_error(400, 'bad url %s' % self.path)
    
37.             raise RuntimeError('bad url %s' % self.path)
    
38.         sock = socket()
    
39.         self._connect_remote(sock, res.netloc)
    
40.         path = urlunparse(('', '', res.path, res.params, res.query, '')) or '/'
    
41.         sock.send('%s %s %s\r\n' % (self.command, path, self.request_version))
    
42.         self.headers['Connection'] = 'close'
    
43.         del self.headers['Proxy-Connection']
    
44.         for header in self.headers:
    
45.             sock.send('%s: %s\r\n' % (header, self.headers[header]))
    
46.         sock.send('\r\n')
    
47.         self._do_proxy(sock)
    
48.         self.connection.close()
    
49.         sock.close()
    
50. 
    
51.     def _connect_remote(self, sock, path):
    
52.         if ':' in path:
    
53.             host, port = path.split(':')
    
54.             port = int(port)
    
55.             remote = host, port
    
56.         else:
    
57.             remote = (path, 80)
    
58.         sock.connect(remote)
    
59. 
    
60.     def _do_proxy(self, sock, timeout=timeout):
    
61.         client, server = self.connection, sock
    
62.         cs = [client, server]
    
63.         while 1:
    
64.             rlist, wlist, xlist = select(cs, [], cs, timeout)
    
65.             if xlist:
    
66.                 who = 'client' if client in xlist else 'server'
    
67.                 raise RuntimeError('exceptional condition from %s' % who)
    
68.             if rlist:
    
69.                 for reader in rlist:
    
70.                     writer = client if reader is server else server
    
71.                     data = reader.recv(bufsize)
    
72.                     if not data:
    
73.                         return
    
74.                     writer.send(data)
    
75.             else:
    
76.                 raise RuntimeError('timeout')
    
77. 
    
78. 
    
79. class ThreadingHTTPServer(ThreadingMixIn, HTTPServer):
    
80.     pass
    
81. 
    
82. 
    
83. def serve(addr, server=ThreadingHTTPServer):
    
84.     print 'Serving HTTP Proxy on %s' % str(addr)
    
85.     server(addr, ProxyHandler).serve_forever()
    
86. 
    
87. 
    
88. def parse_argv(argv=argv):
    
89.     if len(argv) == 1:
    
90.         addr = ('0.0.0.0', 8000)
    
91.     else:
    
92.         arg = argv[1]
    
93.         if arg in ('-h', '--help', '-help', 'help'):
    
94.             show_help(argv[0])
    
95.             exit(1)
    
96.         if ':' in arg:
    
97.             host, port = arg.split(':')
    
98.             addr = (host, int(port))
    
99.         else:
    
100.             addr = ('0.0.0.0', int(arg))
     
101.     return addr
     
102. 
     
103. 
     
104. def show_help(progname=None):
     
105.     progname = progname or argv[0]
     
106.     print 'Usage: %s [[host:]port]'
     
107. 
     
108. if __name__ == '__main__':
     
109.     addr = parse_argv()
     
110.     serve(addr)
     

复制代码

其实我是为了把ssh dynamic forwarding的socks5代理转换成更为通用的http代理，这样可以直接指定http_proxy环境变量来控制程序使用代理的行为

于是又写了个socks2http.py

1. 
   
2. 
   
3. #!/usr/bin/env python
   
4. 
   
5. from sys import argv
   
6. 
   
7. import lightproxy
   
8. import socks
   
9. 
   
10. 
    
11. def init_proxy(socks_host, socks_port):
    
12.     socks.setdefaultproxy(socks.PROXY_TYPE_SOCKS4, socks_host, socks_port)
    
13.     lightproxy.socket = socks.socksocket
    
14. 
    
15. 
    
16. def parse_argv(argv=argv):
    
17.     if len(argv) == 1 or argv[1] in ('-h', '--help', '-help', 'help'):
    
18.         show_help(argv[0])
    
19.         exit(1)
    
20.     arg = argv[1]
    
21.     socks_host, socks_port = arg.split(':')
    
22.     socks_port = int(socks_port)
    
23.     addr = lightproxy.parse_argv([argv[0]] + argv[2:])
    
24.     return (socks_host, socks_port), addr
    
25. 
    
26. 
    
27. def show_help(progname=None):
    
28.     progname = progname or argv[0]
    
29.     print 'Usage: %s socks_host:socks_port [[host:]port]' % progname
    
30. 
    
31. 
    
32. def main():
    
33.     socks_proxy, addr = parse_argv()
    
34.     init_proxy(*socks_proxy)
    
35.     lightproxy.serve(addr)
    
36. 
    
37. if __name__ == '__main__':
    
38.     main()
    
39. 
    

复制代码

后来我把lightproxy.py增加了gevent支持，代码约增加了40行

txdgtwpv

再补一个好玩的，两年前有一天午休时睡不着写的

写的原因在这里https://github.com/lilinux/weida ... /tree/master/fuxing

主要是为了练习命令行progress bar的玩法

偷懒用了datetime的total_seconds，需要使用python2.7
run.py

1. 
   
2. #!/usr/bin/env python
   
3. #-*- coding: utf-8 -*-
   
4. 
   
5. import datetime
   
6. import time
   
7. import sys
   
8. 
   
9. try:
   
10.     from praise import head, tail, interupt, error
    
11. except Exception:
    
12.     head = tail = interupt = error = ''
    
13. 
    
14. begin = datetime.datetime(1949, 10, 1)
    
15. end = datetime.datetime(2049, 10, 1)
    
16. total_seconds = (end - begin).total_seconds()
    
17. 
    
18. def rate(now=None):
    
19.     if not now:
    
20.         now = datetime.datetime.today()
    
21.     return (now - begin).total_seconds() / total_seconds
    
22. 
    
23. def progress():
    
24.     while True:
    
25.         current = rate()
    
26.         if current >= 1:
    
27.             yield 1
    
28.             break
    
29.         yield current
    
30. 
    
31. def bar(length=40, dotted=4, step=2, blank=' ', dot='#'):
    
32.     if not hasattr(bar, 'pos'):
    
33.         bar.pos = 0
    
34.     else:
    
35.         bar.pos = (bar.pos + step) % length
    
36.     return '[%s%s%s]' % (blank*bar.pos, dot*dotted, blank*(length-bar.pos-step))
    
37. 
    
38. def show(interval=0.3, with_bar=bar):
    
39.     print head
    
40.     for r in progress():
    
41.         sys.stdout.write('%s %.10f%%\r'%(bar(), r*100))
    
42.         sys.stdout.flush()
    
43.         time.sleep(interval)
    
44.     sys.stdout.write('%s %.10f%%\r'%(bar(blank='#'), r*100))
    
45.     print '\n', tail
    
46. 
    
47. if __name__ == '__main__':
    
48.     try:
    
49.         show()
    
50.     except KeyboardInterrupt, e:
    
51.         print '\n', interupt
    
52.     except Exception, e:
    
53.         print '\n', error
    
54. 
    

复制代码

praise.py

1. 
   
2. from base64 import decodestring
   
3. import locale
   
4. 
   
5. head = '5Lit5Y2O5rCR5peP5q2j5Zyo6YKj5ZWlLi4u'
   
6. tail = (
   
7. '5Lit5Y2O5rCR5peP5bey6aG65Yip5a6M5oiQ5pmu6YCa5aSN5YW05ZKM5paH6Im65aSN5YW05Lul'
   
8. '5aSW55qE56ys5LiJ56eN5aSN5YW077yM5oSf6LCi6L+Z5Lus5aSa5bm05p2l5L2g5Lus55qE5b+N'
   
9. '6ICQ')
   
10. interupt = '5oKo5oiW6K645LiN57uP5oSP55qE5Li+5Yqo5ouv5pWR5LqG5Lit5Y2O5rCR5peP'
    
11. error = '5aaC6Iul5YaN54qv77yM5Yu/6LCT6KiA5LmL5LiN6aKE5Lmf77yB'
    
12. 
    
13. head = decodestring(head)
    
14. tail = decodestring(tail)
    
15. interupt = decodestring(interupt)
    
16. error = decodestring(error)
    
17. 
    
18. transcode = lambda s: s.decode('utf-8').encode('gbk')
    
19. 
    
20. if locale.getpreferredencoding() != 'UTF-8':
    
21.     head = transcode(head)
    
22.     tail = transcode(tail)
    
23.     interupt = transcode(interupt)
    
24.     error = transcode(error)
    

复制代码

后来我把这个脚本嵌进了tmux的状态栏，让命令行的右下角无时无刻地提醒我的伟大使命

ssfjhh

已收到书，待我下班以后细看。:wink:


除了年底的年终奖，已经多少年没有收到过奖品了，激动。

Gubuntu

1. # -*- coding: utf-8 -*-
   
2. import xml.dom.minidom
   
3. ELEMENT_NODE = xml.dom.Node.ELEMENT_NODE
   
4. 
   
5. class SimpleXmlGetter(object):
   
6.     def __init__(self, data):
   
7.         if type(data) == str:
   
8.             self.root = xml.dom.minidom.parse(data)
   
9.         else:
   
10.             self.root = data
    
11. 
    
12.     def __getattr__(self, name):        #support . operation
    
13.         if name == 'data':
    
14.             return self.root.firstChild.data
    
15.         for c in self.root.childNodes:
    
16.             if c.nodeType == ELEMENT_NODE and c.tagName == name:
    
17.                 return SimpleXmlGetter(c)
    
18. 
    
19.     def __getitem__(self, index):       #support [] operation
    
20.         eNodes = [ e for e in self.root.parentNode.childNodes
    
21.                     if e.nodeType == ELEMENT_NODE and e.tagName == self.root.tagName]
    
22.         return SimpleXmlGetter(eNodes[index])
    
23. 
    
24.     def __call__(self, *args, **kwargs):  #support () openration, for query conditions
    
25.         for e in self.root.parentNode.childNodes:
    
26.             if e.nodeType == ELEMENT_NODE:
    
27.                 for key in kwargs.keys():
    
28.                      if e.getAttribute(key) != kwargs[key]:
    
29.                         break
    
30.                 else:
    
31.                     return SimpleXmlGetter(e)
    
32. 
    
33. if __name__ == "__main__":
    
34.     x = SimpleXmlGetter("test.xml")
    
35.     print x.sysd.sysagent.param[2].data
    
36.     print x.sysd.sysagent.param(name="querytimeout", type="second").data

复制代码

获取xml中的某一节点的值，支持通过.取子节点，[]索引，()条件查询