Mechanize实现网站404监控

琴俊人

2023-12-01

【前言】

网站的404监控，其实要用到的就urllib2和re这两个模块，urllib2用来处理请求，re正则表达式用来处理html页面。

其中，处理html页面（找出页面的全部links），也可使用BeautifulSoup，简单方便。

这里，将介绍如何使用mechanize模块实现网站的404监控。主要内容包括：

1、实现过程中常见的问题答疑；

2、多线程中的线程安全问题；

mechanize模块，将urllib2和beautifuSoup封装起来，同时模拟浏览器的操作。

对于web测试工具开发，mechanize是个不错的辅助工具。

就是mechanize的帮助文档太少，同时FAQ的内容不多，完完全全需要你自己是参考他的源码。

codesearch：https://searchcode.com/codesearch/view/80174766/

源码：http://www.joesourcecode.com/Documentation/mechanize0.2.5/

【需求分析】

网站404监控：

1、监控全网所有links

隐性需求：

1、很多网站都有会员登录，links会有所不同（使用cookie登录即可）；

2、同时不同角色会员登录，links会有所不同；

3、每个link下还有其他链接（可使用递归处理，相当于爬虫工具，难点在于如何终止递归）；

4、像功能链接，很容易被忽略，如【下一页】，它的href会像 /page/1 这种格式；

以上这些隐性需求，本文暂不考虑，写起来也不难，你可以用来练练手，修改下面的代码。

【问题解决】

1、有些网站有太多链接，如果串行访问，导致一个脚本跑下来需要花费，如果你机器差的话，那就更糟糕。

解决方法：使用thread模块，创建多进程并发访问，如一个主要页面一个进程，每个进程串行访问url时停顿1秒。

请注意，有些网站拒绝短时间内发送多个请求的链接，或者拒绝短时间内多次请求同一个链接，所以停顿时间需要设置合适。

2、网站404监控完成，如果url访问存在【code != 200】（返回码）的话，要发送邮件，贴上错误的url（记录在相应的txt文件中）。

解决方案：此时已采用了多线程，如何去汇总所有错误url，且该如何合理触发邮件呢？这就想到了我之前提过的装饰器。

我添加个装饰器，使用一个全局变量列表，记录所有错误的url，代码如下：

def monitoring(files):
    '''用来监控线程，每次访问url时，是否有code!=200的，如果是，则添加文件（记录着错误的url）
    '''
    def wrap(f):
        def newFunction(*args,**kw):
            result = f(*args,**kw)
            if result[0]:
                files.append(result[1])
        return newFunction
    return wrap

上面的装饰器用来装饰线程实例的run函数，如下：

send_files = []    
    @monitoring(send_files)
    def do_run(self):
        '''每个进程执行的内容，多进程进行three_test_404
        '''
        links_dict = self.get_sub_links_dict()
        urls = links_dict.keys()
        shuffle(urls) #打乱每个进程访问的urls列表，因为进程间访问的url有重复的风险，这里尽量避免同时访问同一个url
        errcount = 0
        for url in urls:
            if not url.startswith('javascript'):
                code = three_test_404(self._br,url)
                if not code or 200 != int(code):
                    errcount += 1
                    self._output.writelines(url+ '  code: ' + str(code)+' title:'+ links_dict[url].text + '\n')

        self._output.writelines('\n[' + re.sub('\n','',self._url) + '] errors:  ' + str(errcount) + '\n') 
        filename = self._output.name
        self._output.close()
        
        return (errcount,filename)

3、获取某个web页面所有链接的有两种方法：

（1）使用mechanize.Browser()，如下：

br = mechanize.Browser()

br.open(weburlA)

links = br.links()

请注意，links是一个迭代器，请不要在links迭代中使用br.open(weburlB)访问另一个url，否则links将成为weburB下的所有链接。

（2）使用mechanize.LinksFactory()，推荐使用，如下：

lf = mechanize.LinksFactory()

response = urllib2.open(weburlA)

lf.set_response(response)

links = lf.links()

在获取页面所有链接links后，建议将其转换为列表，以免后续不小心使用【lf.set_response()】或者【br.open()】，导致程序结果有误。

另外，在使用links迭代时，如下：

for link in links():
     do somethting

很容易出现（socket error）10054错误，除非你运气极好，具体信息如下：

socket.error: [Errno 10054] An existing connection was forcibly closed by the remote host

个人认为，这算mechanize的一个bug，links迭代的过程中，mechanize不断使用response.read(1024)读取页面html。

这里的response，是之前你使用br.open()打开的一个请求（file-like object）返回。

解决方案：Browser对象或者LinksFactory对象，都提供一个函数set_response()，让你设置一个（file-like object，即是有read函数的实例）请求返回。

所以，这里可以将br.open()返回response的html内容，存入临时文件tempfile，然后使用tempfile的文件句柄作为response即可。

        response = self.get_response(self._url, 10)
        temp = tempfile.TemporaryFile()
        temp.write(response.read())
        temp.seek(0)

        if not response:
            raise RuntimeError('[%s] open failed' %self._url)
        self._lf.set_response(temp,self._url,'utf-8')
        links_dict = {}
        
        for link in self._lf.links():
            #当字典key有该url，且此时link无text内容，则不需要替换
            if links_dict.has_key(link.url) and not link.text:
                continue
            links_dict[link.url] = link
            
        temp.close()

【代码部分解析】

MultiTest404.py：检测404，这里选取三次测试，如果有一次为200，则成功，这样能减少偶然事件

def test_404(browser,url):
    code = None
    try:
        response = browser.open(url)
        #response = browser.response()#发送请求
        code = response.code
    except HTTPError,e:
        code = e.code #获取http错误码        
    except (URLError,error),e:
        try:
            code = str(e.args[0].errno) #获取socket的返回码，errorcode 是字典
        except:
            code = 10000 #获取不到socket错误码，自定义为1000
    except:
        code = 10001 #其他错误，统一自定义为1001
        
    return code

def three_test_404(browser,url,wait=8):
    '''三次测试：如果有一次返回码为200，则认为成功，返回200，否则取最后一次的返回码
    '''
    for i in xrange(3):
        code = test_404(browser,url)
        if 200 == int(code):
            break
        time.sleep(wait)

    return code

MultiTest404.py：线程主体，注意： 这里使用一个browser，一个linksfactory

def synchronized(lock):
    """ Synchronization decorator.
             同步锁，用来锁住争抢资源
    """
    def wrap(f):
        def newFunction(*args,**kw):
            lock.acquire()
            try:
                return f(*args,**kw)
            finally:
                lock.release()
        return newFunction
    return wrap

class urlThread(Thread):
    
    def __init__(self,browser,link_factory,url):
        Thread.__init__(self)
        self._url = url
        self._br = browser
        self._lf = link_factory
        name = '(\w*)\.letv'
        ch1 = re.findall(name,url)
        fname=control.get_file(str(ch1[0]))
        self._output = open(fname,'w')
    
    @synchronized(mylock)    
    def get_sub_links_dict(self):
        '''获取links字典，url为key'''
        response = self.get_response(self._url, 10)
        temp = tempfile.TemporaryFile()
        temp.write(response.read())
        temp.seek(0)

        if not response:
            raise RuntimeError('[%s] open failed' %self._url)
        self._lf.set_response(temp,self._url,'utf-8')
        links_dict = {}
        
        for link in self._lf.links():
            #当字典key有该url，且此时link无text内容，则不需要替换
            if links_dict.has_key(link.url) and not link.text:
                continue
            links_dict[link.url] = link
            
        temp.close()

        return links_dict
            
    def get_response(self,url,times=3,wait=10):
        '''生成每个主要页面的请求，并获取返回值，如果这个过程出错，则等待wait秒，次数为times
        '''
        num = times
        response = None

        while num:
            try:
                response = urlopen(url)
                break
            except:
                time.sleep(wait)
                num -= 1
                
        return response
    
    @monitoring(send_files)
    def run(self):
        '''每个进程执行的内容，多进程进行three_test_404
        '''
        #2015-01-22 修改，提高效率
        links_dict = self.get_sub_links_dict()
        urls = links_dict.keys()
        shuffle(urls) #打乱每个进程访问的urls列表，因为进程间访问的url有重复的风险，这里尽量避免同时访问同一个url
        errcount = 0
        for url in urls:
            if not url.startswith('javascript'):
                code = three_test_404(self._br,url)
                if not code or 200 != int(code):
                    errcount += 1
                    self._output.writelines(url+ '  code: ' + str(code)+' title:'+ links_dict[url].text + '\n')

        self._output.writelines('\n[' + re.sub('\n','',self._url) + '] errors:  ' + str(errcount) + '\n') 
        filename = self._output.name
        self._output.close()
        
        return (errcount,filename)

以上有两个主要点，涉及【 线程安全】问题：
1、串行获取每个主要页面的links，这里使用同步锁锁着公共资源self._lf，线程安全；
2、并行进行主要页面links的404监控，没有使用同步锁，每个线程同时使用公共资源self._br；
至于2，为什么不线程安全呢？请test_404函数代码：

        response = browser.open(url) #这里线程不安全
        #response = browser.response()#发送请求
        code = response.code

如上， 7个线程，某一时刻，如果其中一个线程A刚好使用browser打开urlA，没等到A去获取code时，线程B又去打开另外一个urlB。
那线程A获取的code却成了urlB的返回码，导致结果有出入。
显然，整个过程会乱套。你可以将以上几句换成一句，如下：

code = browser.open(url).code

但还是不安全，只是减少出错的概率。

这时，你会质疑，这个程序还有用么？

个人认为，虽然线程不安全，但是由于请求的时间极短，出错的概率很低。

同时，整个网站返回码不是200,的也就那几个，即使我乱配鸳鸯，又能错了多少缘分。

要解决上面的【线程安全】问题，也很简单，就是多开辟几个Browser和linksFactory，即每个线程不存在共享资源。

更简单点，就是直接给run函数上锁（@synchronized(mylock) ），但这样就成了串行。

还有一个解决方案，使用【线程池】，而且速度能提升1/3以上。

对上面所说，如有什么问题，可以留言。

Mechanize实现网站404监控

【前言】

【需求分析】

【问题解决】

【代码部分解析】

相关阅读

相关文章

相关问答

相关文档