免费注册 查看新帖 |

Chinaunix

  平台 论坛 博客 文库
1234
最近访问板块 发新帖
楼主: timespace

分享有礼:Python代码的魔鬼细节大比拼!(获奖名单已公布-2014-8-11) [复制链接]

论坛徽章:
19
处女座
日期:2014-07-18 14:50:5415-16赛季CBA联赛之北京
日期:2019-09-16 15:39:1415-16赛季CBA联赛之上海
日期:2019-09-15 15:29:0415-16赛季CBA联赛之山西
日期:2017-03-09 10:58:232017金鸡报晓
日期:2017-02-08 10:33:212017金鸡报晓
日期:2017-01-10 15:13:2915-16赛季CBA联赛之深圳
日期:2016-12-15 17:55:53C
日期:2016-10-25 16:00:1515-16赛季CBA联赛之新疆
日期:2016-07-21 14:02:0415-16赛季CBA联赛之江苏
日期:2016-06-30 12:15:04shanzhi
日期:2016-06-17 17:59:31平安夜徽章
日期:2015-12-26 00:06:30
发表于 2014-07-26 22:12 |显示全部楼层
对Python有一点了解,LP在工作中用得比较多,支持活动,顶一个。

论坛徽章:
13
午马
日期:2015-01-19 14:08:552017金鸡报晓
日期:2017-01-10 15:13:29黑曼巴
日期:2016-11-07 11:24:56PHP
日期:2016-10-25 16:06:46黄金圣斗士
日期:2015-11-24 10:43:13IT运维版块每日发帖之星
日期:2015-09-25 06:20:00IT运维版块每日发帖之星
日期:2015-09-14 06:20:002015亚冠之阿尔纳斯尔
日期:2015-07-27 11:17:582015亚冠之广州恒大
日期:2015-07-24 15:04:162015年亚洲杯之乌兹别克斯坦
日期:2015-04-01 13:28:012015年辞旧岁徽章
日期:2015-03-03 16:54:15处女座
日期:2015-01-22 16:09:16
发表于 2014-07-30 16:41 |显示全部楼层
好活动  必须顶啊,对python只是偶尔能简单用点

论坛徽章:
2
天蝎座
日期:2014-03-28 10:18:052015年亚洲杯之乌兹别克斯坦
日期:2015-02-10 11:32:25
发表于 2014-08-01 17:08 |显示全部楼层
前几天写的HTTP代理服务器  lightproxy.py, 除掉空行在100行以内

  1. #!/usr/bin/env python
  2. from select import select
  3. from socket import socket
  4. from sys import argv, exit
  5. from urlparse import urlparse, urlunparse
  6. from BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer
  7. from SocketServer import ThreadingMixIn

  8. __version__ = '0.01'
  9. bufsize = 1024 * 1024  # 1MB
  10. timeout = 60


  11. class ProxyHandler(BaseHTTPRequestHandler):
  12.     server_version = 'SimpleProxy/%s' % __version__

  13.     def do_CONNECT(self):
  14.         self.log_request()
  15.         sock = socket()
  16.         self._connect_remote(sock, self.path)
  17.         self.wfile.write('%s 200 Connection Established\r\n'
  18.                          % self.protocol_version)
  19.         self.wfile.write('Proxy-agent: %s\r\n' % self.version_string())
  20.         self.wfile.write('\r\n')
  21.         self._do_proxy(sock)
  22.         self.connection.close()
  23.         sock.close()

  24.     def do_GET(self):
  25.         import time
  26.         print time.ctime()
  27.         self.log_request()
  28.         res = urlparse(self.path)
  29.         if res.scheme != 'http' or res.fragment or not res.netloc:
  30.             self.send_error(400, 'bad url %s' % self.path)
  31.             raise RuntimeError('bad url %s' % self.path)
  32.         sock = socket()
  33.         self._connect_remote(sock, res.netloc)
  34.         path = urlunparse(('', '', res.path, res.params, res.query, '')) or '/'
  35.         sock.send('%s %s %s\r\n' % (self.command, path, self.request_version))
  36.         self.headers['Connection'] = 'close'
  37.         del self.headers['Proxy-Connection']
  38.         for header in self.headers:
  39.             sock.send('%s: %s\r\n' % (header, self.headers[header]))
  40.         sock.send('\r\n')
  41.         self._do_proxy(sock)
  42.         self.connection.close()
  43.         sock.close()

  44.     def _connect_remote(self, sock, path):
  45.         if ':' in path:
  46.             host, port = path.split(':')
  47.             port = int(port)
  48.             remote = host, port
  49.         else:
  50.             remote = (path, 80)
  51.         sock.connect(remote)

  52.     def _do_proxy(self, sock, timeout=timeout):
  53.         client, server = self.connection, sock
  54.         cs = [client, server]
  55.         while 1:
  56.             rlist, wlist, xlist = select(cs, [], cs, timeout)
  57.             if xlist:
  58.                 who = 'client' if client in xlist else 'server'
  59.                 raise RuntimeError('exceptional condition from %s' % who)
  60.             if rlist:
  61.                 for reader in rlist:
  62.                     writer = client if reader is server else server
  63.                     data = reader.recv(bufsize)
  64.                     if not data:
  65.                         return
  66.                     writer.send(data)
  67.             else:
  68.                 raise RuntimeError('timeout')


  69. class ThreadingHTTPServer(ThreadingMixIn, HTTPServer):
  70.     pass


  71. def serve(addr, server=ThreadingHTTPServer):
  72.     print 'Serving HTTP Proxy on %s' % str(addr)
  73.     server(addr, ProxyHandler).serve_forever()


  74. def parse_argv(argv=argv):
  75.     if len(argv) == 1:
  76.         addr = ('0.0.0.0', 8000)
  77.     else:
  78.         arg = argv[1]
  79.         if arg in ('-h', '--help', '-help', 'help'):
  80.             show_help(argv[0])
  81.             exit(1)
  82.         if ':' in arg:
  83.             host, port = arg.split(':')
  84.             addr = (host, int(port))
  85.         else:
  86.             addr = ('0.0.0.0', int(arg))
  87.     return addr


  88. def show_help(progname=None):
  89.     progname = progname or argv[0]
  90.     print 'Usage: %s [[host:]port]'

  91. if __name__ == '__main__':
  92.     addr = parse_argv()
  93.     serve(addr)
复制代码
其实我是为了把ssh dynamic forwarding的socks5代理转换成更为通用的http代理,这样可以直接指定http_proxy环境变量来控制程序使用代理的行为

于是又写了个socks2http.py


  1. #!/usr/bin/env python

  2. from sys import argv

  3. import lightproxy
  4. import socks


  5. def init_proxy(socks_host, socks_port):
  6.     socks.setdefaultproxy(socks.PROXY_TYPE_SOCKS4, socks_host, socks_port)
  7.     lightproxy.socket = socks.socksocket


  8. def parse_argv(argv=argv):
  9.     if len(argv) == 1 or argv[1] in ('-h', '--help', '-help', 'help'):
  10.         show_help(argv[0])
  11.         exit(1)
  12.     arg = argv[1]
  13.     socks_host, socks_port = arg.split(':')
  14.     socks_port = int(socks_port)
  15.     addr = lightproxy.parse_argv([argv[0]] + argv[2:])
  16.     return (socks_host, socks_port), addr


  17. def show_help(progname=None):
  18.     progname = progname or argv[0]
  19.     print 'Usage: %s socks_host:socks_port [[host:]port]' % progname


  20. def main():
  21.     socks_proxy, addr = parse_argv()
  22.     init_proxy(*socks_proxy)
  23.     lightproxy.serve(addr)

  24. if __name__ == '__main__':
  25.     main()

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

论坛徽章:
2
天蝎座
日期:2014-03-28 10:18:052015年亚洲杯之乌兹别克斯坦
日期:2015-02-10 11:32:25
发表于 2014-08-01 17:30 |显示全部楼层
本帖最后由 txdgtwpv 于 2014-08-01 17:31 编辑

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

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

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

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

  1. #!/usr/bin/env python
  2. #-*- coding: utf-8 -*-

  3. import datetime
  4. import time
  5. import sys

  6. try:
  7.     from praise import head, tail, interupt, error
  8. except Exception:
  9.     head = tail = interupt = error = ''

  10. begin = datetime.datetime(1949, 10, 1)
  11. end = datetime.datetime(2049, 10, 1)
  12. total_seconds = (end - begin).total_seconds()

  13. def rate(now=None):
  14.     if not now:
  15.         now = datetime.datetime.today()
  16.     return (now - begin).total_seconds() / total_seconds

  17. def progress():
  18.     while True:
  19.         current = rate()
  20.         if current >= 1:
  21.             yield 1
  22.             break
  23.         yield current

  24. def bar(length=40, dotted=4, step=2, blank=' ', dot='#'):
  25.     if not hasattr(bar, 'pos'):
  26.         bar.pos = 0
  27.     else:
  28.         bar.pos = (bar.pos + step) % length
  29.     return '[%s%s%s]' % (blank*bar.pos, dot*dotted, blank*(length-bar.pos-step))

  30. def show(interval=0.3, with_bar=bar):
  31.     print head
  32.     for r in progress():
  33.         sys.stdout.write('%s %.10f%%\r'%(bar(), r*100))
  34.         sys.stdout.flush()
  35.         time.sleep(interval)
  36.     sys.stdout.write('%s %.10f%%\r'%(bar(blank='#'), r*100))
  37.     print '\n', tail

  38. if __name__ == '__main__':
  39.     try:
  40.         show()
  41.     except KeyboardInterrupt, e:
  42.         print '\n', interupt
  43.     except Exception, e:
  44.         print '\n', error

复制代码
praise.py

  1. from base64 import decodestring
  2. import locale

  3. head = '5Lit5Y2O5rCR5peP5q2j5Zyo6YKj5ZWlLi4u'
  4. tail = (
  5. '5Lit5Y2O5rCR5peP5bey6aG65Yip5a6M5oiQ5pmu6YCa5aSN5YW05ZKM5paH6Im65aSN5YW05Lul'
  6. '5aSW55qE56ys5LiJ56eN5aSN5YW077yM5oSf6LCi6L+Z5Lus5aSa5bm05p2l5L2g5Lus55qE5b+N'
  7. '6ICQ')
  8. interupt = '5oKo5oiW6K645LiN57uP5oSP55qE5Li+5Yqo5ouv5pWR5LqG5Lit5Y2O5rCR5peP'
  9. error = '5aaC6Iul5YaN54qv77yM5Yu/6LCT6KiA5LmL5LiN6aKE5Lmf77yB'

  10. head = decodestring(head)
  11. tail = decodestring(tail)
  12. interupt = decodestring(interupt)
  13. error = decodestring(error)

  14. transcode = lambda s: s.decode('utf-8').encode('gbk')

  15. if locale.getpreferredencoding() != 'UTF-8':
  16.     head = transcode(head)
  17.     tail = transcode(tail)
  18.     interupt = transcode(interupt)
  19.     error = transcode(error)
复制代码
后来我把这个脚本嵌进了tmux的状态栏,让命令行的右下角无时无刻地提醒我的伟大使命

论坛徽章:
4
金牛座
日期:2013-10-11 16:12:50卯兔
日期:2014-07-31 09:17:19辰龙
日期:2014-08-08 09:28:02狮子座
日期:2014-09-14 20:32:05
发表于 2014-08-15 11:34 |显示全部楼层
已收到书,待我下班以后细看。:wink:


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

论坛徽章:
1
巨蟹座
日期:2014-08-15 14:56:13
发表于 2014-08-15 15:18 |显示全部楼层
  1. # -*- coding: utf-8 -*-
  2. import xml.dom.minidom
  3. ELEMENT_NODE = xml.dom.Node.ELEMENT_NODE

  4. class SimpleXmlGetter(object):
  5.     def __init__(self, data):
  6.         if type(data) == str:
  7.             self.root = xml.dom.minidom.parse(data)
  8.         else:
  9.             self.root = data

  10.     def __getattr__(self, name):        #support . operation
  11.         if name == 'data':
  12.             return self.root.firstChild.data
  13.         for c in self.root.childNodes:
  14.             if c.nodeType == ELEMENT_NODE and c.tagName == name:
  15.                 return SimpleXmlGetter(c)

  16.     def __getitem__(self, index):       #support [] operation
  17.         eNodes = [ e for e in self.root.parentNode.childNodes
  18.                     if e.nodeType == ELEMENT_NODE and e.tagName == self.root.tagName]
  19.         return SimpleXmlGetter(eNodes[index])

  20.     def __call__(self, *args, **kwargs):  #support () openration, for query conditions
  21.         for e in self.root.parentNode.childNodes:
  22.             if e.nodeType == ELEMENT_NODE:
  23.                 for key in kwargs.keys():
  24.                      if e.getAttribute(key) != kwargs[key]:
  25.                         break
  26.                 else:
  27.                     return SimpleXmlGetter(e)

  28. if __name__ == "__main__":
  29.     x = SimpleXmlGetter("test.xml")
  30.     print x.sysd.sysagent.param[2].data
  31.     print x.sysd.sysagent.param(name="querytimeout", type="second").data
复制代码
获取xml中的某一节点的值,支持通过.取子节点,[]索引,()条件查询
您需要登录后才可以回帖 登录 | 注册

本版积分规则 发表回复

DTCC2020中国数据库技术大会

【架构革新 高效可控】2020年12月21日-23日第十一届中国数据库技术大会将在北京隆重召开。

大会设置2大主会场,20+技术专场,将邀请超百位行业专家,重点围绕数据架构、AI与大数据、传统企业数据库实践和国产开源数据库等内容展开分享和探讨,为广大数据领域从业人士提供一场年度盛会和交流平台。

http://dtcc.it168.com


大会官网>>
  

北京盛拓优讯信息技术有限公司. 版权所有 16024965号-6 北京市公安局海淀分局网监中心备案编号:11010802020122
中国互联网协会会员  联系我们:huangweiwei@it168.com
感谢所有关心和支持过ChinaUnix的朋友们 转载本站内容请注明原作者名及出处

清除 Cookies - ChinaUnix - Archiver - WAP - TOP