- 灵感点滴

让 ActivePython 支持 SSL

由于受加密出口的限制, ActivePython的安装包中没有SSL支持, 导致urllib2等模块不能支持HTTPS协议, 在官方网站上有说明: Why doesn't ActivePython support SSL?

文中提到, Python.org上提供Python安装包中有SSL模块, 将安装目录中的DLLs/_ssl.pyd文件拷贝到ActivePython安装目录的DLLs目录下即可。

另外，Python中的代理目前不支持HTTPS协议，可以找到一个补丁(Google 之)，但是实现得不太好，不推荐。

davies 发表于 2006 年 11 月 16 日 | 0 条留言

镜像山野的网站

山野的网站是用MSSQL+ASP做的，放在校外某公司的服务器上，访问起来很慢，维护很不方便。为了提高访问速度，想给它做一个镜像，用程序把网站里面的所有资源爬下来保存为静态页面，并按照原来的结构组织好，做为一个完全静态的网站。

Linux下的wget是镜像网站的好工具，它的-mirror参数就是专门干这个的。能够根据页面的修改时间来选择是否下载页面。而且wget能够用完成URL作为文件名，使得具有不同参数的动态内容保存为不同的静态页面成为可能。但wget有一个致命缺陷，它不支持多线程，本来是因为访问原站点很慢才想要做镜像，用单线程的话会使镜像过程相当漫长，变得不太可能。

网上有不少现成的爬虫工具，用Python实现的就有好几个，试了一下，都有一个通病：保存文件时不带参数，不行。懒得去hack它的程序了，还不如自己用Python写一个，其实也蛮简单的，无非是用多线程去下载页面，这个利用前面已经完成的线程池Worker.py将相当简单。其它就是解析HTML页面，找出其中的链接，而山野的网站有其自身的特点，写一个适用于它而并非通用的解析器也是很简单的。

保存的时候，需要根据URL生成文件路径，将其中的?替换成_，并在最后加一个.html后缀。这样的话，这些镜像后的静态文件的URL跟页面中的内部URL会不一样，一种方法就是改变页面中的链接为静态文件的链接，这是浏览器下载页面通常采用的方法，而另一种更简单的方法是利用HTTP服务器的rewrite功能，用与转换文件名同样的规则来作为rewrite规则。这样对访问者来讲，链接及内容都没有改变，而实际上它已经是一个个的静态HTML页面了。上面给出链接就是镜像后的效果，而原站点在这里：http://219.239.7.49/braveheart/

另外，为了避免循环下载，需要保存一个已经开始下载的URL列表，如果在页面中找到的新的URL已经在这个列表中，则无需下载，否则会死循环。

完整的代码如下，可以在这里下载。

#!/usr/bin/python
import re,urllib,os
import worker, Queue

def link2filename(link):
    if link.count('.asp?'):
        return link.replace('?','_')+'.html'
    else:
        return link

def find_links( filename ):
    base = ''
    if filename.find('/') > 0: base = filename[:filename.index('/')+1]
    content = open(filename).read()
    href = re.findall(r'href="(.*?)"', content)
    src = re.findall(r'src="(.*?)"', content)
    urls = href + src
    urls = filter(lambda s:not s.startswith('http'), urls)
    urls = filter(lambda s:not s.startswith('mailto'), urls)
    urls = filter(lambda s:not s.startswith('javascript'), urls)
    res = []
    for url in urls:
        if url.startswith('../'):
            res.append(url[3:])
        else:
            res.append(base+url)
    return res


def download(link):
    urllib.urlretrieve('http://219.239.7.49/braveheart/'+link, link2filename(link))
    return link

def mirror(url):
    links = open('urls.log').read().split('\n')
    os.chdir('braveheart')
    worker.Worker.timeout = None
    wm = worker.WorkerManager(50)
    wm.add_job(download, url)
    while True:
        try:
            link = wm.get_result(True, timeout = 90)
            print link
            links.append(link)
            filename = link2filename(link)
            if filename.endswith('html') > 0:
                new_links = find_links( filename )
                new_links = [l for l in new_links if l not in links]
                for l in new_links:
                    wm.add_job(download, l)
        except Queue.Empty:
            break
    open('../urls.log','w').write( '\n'.join(links) )

if __name__ == '__main__':
    mirror('showcatalog3.asp?CatalogID=1')

davies 发表于 2006 年 10 月 12 日 | 2 条留言

用 Python 实现的线程池

为了提高程序的效率，经常要用到多线程，尤其是IO等需要等待外部响应的部分。线程的创建、销毁和调度本身是有代价的，如果一个线程的任务相对简单，那这些时间和空间开销就不容忽视了，此时用线程池就是更好的选择，即创建一些线程然后反复利用它们，而不是在完成单个任务后就结束。

下面是用Python实现的通用的线程池代码：

import Queue, threading, sys
from threading import Thread
import time,urllib

# working thread
class Worker(Thread):
	worker_count = 0
	def __init__( self, workQueue, resultQueue, timeout = 0, **kwds):
		Thread.__init__( self, **kwds )
		self.id = Worker.worker_count
		Worker.worker_count += 1
		self.setDaemon( True )
		self.workQueue = workQueue
		self.resultQueue = resultQueue
		self.timeout = timeout

	def run( self ):
		''' the get-some-work, do-some-work main loop of worker threads '''
		while True:
			try:
				callable, args, kwds = self.workQueue.get(timeout=self.timeout)
				res = callable(*args, **kwds)
				print "worker[%2d]: %s" % (self.id, str(res) )
				self.resultQueue.put( res )
			except Queue.Empty:
				break
			except :
				print 'worker[%2d]' % self.id, sys.exc_info()[:2]
				
class WorkerManager:
	def __init__( self, num_of_workers=10, timeout = 1):
		self.workQueue = Queue.Queue()
		self.resultQueue = Queue.Queue()
		self.workers = []
		self.timeout = timeout
		self._recruitThreads( num_of_workers )

	def _recruitThreads( self, num_of_workers ):
		for i in range( num_of_workers ):
			worker = Worker( self.workQueue, self.resultQueue, self.timeout )
			self.workers.append(worker)

	def start(self):
		for w in self.workers:
			w.start()

	def wait_for_complete( self):
		# ...then, wait for each of them to terminate:
		while len(self.workers):
			worker = self.workers.pop()
			worker.join( )
			if worker.isAlive() and not self.workQueue.empty():
				self.workers.append( worker )
		print "All jobs are are completed."

	def add_job( self, callable, *args, **kwds ):
		self.workQueue.put( (callable, args, kwds) )

	def get_result( self, *args, **kwds ):
		return self.resultQueue.get( *args, **kwds )

Worker类是一个工作线程，不断地从workQueue队列中获取需要执行的任务，执行之，并将结果写入到resultQueue中，这里的workQueue和resultQueue都是现成安全的，其内部对各个线程的操作做了互斥。当从workQueue中获取任务超时，则线程结束。

WorkerManager负责初始化Worker线程，提供将任务加入队列和获取结果的接口，并能等待所有任务完成。

一个典型的测试例子如下，它用10个线程去下载一个固定页面的内容，实际应用时应该是执行不同的任务。

def test_job(id, sleep = 0.001 ):
	try:
		urllib.urlopen('https://www.gmail.com/').read()
	except:
		print '[%4d]' % id, sys.exc_info()[:2]
	return  id

def test():
	import socket
	socket.setdefaulttimeout(10)
	print 'start testing'
	wm = WorkerManager(10)
	for i in range(500):
		wm.add_job( test_job, i, i*0.001 )
	wm.start()
	wm.wait_for_complete()
	print 'end testing'

完成的程序可以在这里下载。

davies 发表于 2006 年 10 月 9 日 | 13 条留言

PyCAPTCHA 的 BUG

这两天在使用PyCAPTCHA模块的过程中发现了它的一个bug：使用PersistentFactory 来管理CAPTCHA测试用例时，一个测试用例可以多次测试成功而不会实效，这样可以人工得到答案后反复提交该答案而绕过CAPTCHA测试。

在PyCAPTCHA的已知Bug列表中已经有类似的问题：

PersistentFactory() is almost certainly horrible at concurrent access
Tests are never invalidated with PersistentStorage(), as they aren't written back to the database
All files in Captcha/data are installed, including silly things like .svn directories and *~

这是因为PersistentFactory使用shelve来存储测试用例，已达到进程间共享数据的目的。而它使用shelve打开文件时，默认参数writeback为False，即关闭文件时必会自动回写。如果数据在打开后的修改操作不是通过顶层的字典来进行，该修改不会被shelve发现，也不会写回到文件中，从而出现上面的问题。这个问题在shelve模块的例子中已经明确注明。

一种解决方案是在打开文件时加 writeback=True参数，每次关闭文件时都会将数据内容写回到文件中，而不管是否作了修改，会比较慢。

另一种解决方案是每次修改数据内容时都显示的进行，比如：

采用这种方法，修改BaseFactory的test()函数为：

    def test(self, id, solutions):
        """Test the given list of solutions against the BaseCaptcha instance
           created earlier with the given id. Returns True if the test passed,
           False on failure. In either case, the test is invalidated. Returns
           False in the case of an invalid id.
           """
        self.clean()
        inst = self.storedInstances.get(id)
        if not inst:
            return False
        result = inst.testSolutions(solutions)
        # write back modification of inst
        self.storedInstances[id] = inst
        return result

这样在调用PersistentFactory的test()函数后就会将修改后的test内容写回到文件中，使该测试用例实效。但这种方法还存在一个问题，如果直接调用Captcha对象的testSolutions方法，仍然会存在这个问题。

本质上这是由PersistentFactory所用的shelve模块的不完备导致的，同时由于shelve模块不是多进程安全的，导致了已知BUG列表中的第一个。要从根本上解决问题，只能换用其他的序列化方法，具有多进程多线程安全并且自动回写的特点。

davies 发表于 2006 年 09 月 17 日 | 0 条留言

过程式编程与函数式编程

过程式与函数式是两种截然不同的编程方式和思考方法，下面以求解素数为例做一下对比。

采用函数式编程如下：

而如果采用函数式，则可以这样写：

它同上面的算法是一样的，想看懂的话必须先知道map、reduce的用法，参考Python的官方文档，提示一下：l表示已经找到的素数序列，not 0 in map(lambda x:y %x,l) 表示数y能否被l中的任何一个数整除，继而返回l+[y]或者l。

对比一下这两段程序，可以明显地看出过程式的代码虽长但直白，适合初学算法的人，而函数式的代码短而晦涩，有着数学一样的抽象，适合hacker。但是如果你习惯了函数编程的思维方式，反而会觉得代码直观明了。不管采用哪种方式的编程，代码的可读性都是非常重要的，要根据具体的场合选用合适的编程方式。

从效率的角度讲，一般函数式编程的效率会低一些。比如上面的例子，在确定一个数是不是质数时，过程式只要找到了一个它的因数就返回，而函数式需要除以比它小的所有质数，计算量要多一些。

davies 发表于 2006 年 03 月 5 日 | 4 条留言

上一页 | 第 2 / 6 页 | 下一页

灵感点滴

让 ActivePython 支持 SSL

镜像山野的网站

用 Python 实现的线程池

PyCAPTCHA 的 BUG

过程式编程与函数式编程

搜索文章

最新文章

历史存档

文章分类

订阅