多线程校验url的种种。。。

Posted 2020-06-14

东西不经常看就容易忘，这两天抽空写了一个补丁库url验证工具，挺曲折的，记录一下
需求大概流程如下：转换补丁库为xml，解析xml得到所有url，多线程校验url。
就不到100行有效代码，断断续续写了三天才完成：
1、调用dll，忘记了参数类型对应的ctype类型
2、解析xml，没有搞清楚需求，以为要解析json，所以写了一个解析json的函数，最后发现没用上。中间解析json、xml的方法又看了一下资料才能拣起来
3、多线程，忘记传参数的方法
4、校验url，本来自己写了几种方法，结果发现还是老外的方法好一点。。。

def anyTrue(a, b):
    return True in map(a, b)
        
def file_exists1(url,folder):
    for i in range(len(url)): 
        request = urllib2.Request(str(url[i]))
        request.get_method = lambda : ‘HEAD‘
        try:
            response = urllib2.urlopen(request)
            continue
        except Exception,info:
            # print info
            ff = open(folder+ ‘/‘ + ‘fail.txt‘,‘a‘)
            ff.write(url[i]+‘----‘+str(info)+‘\n‘)
            continue
def filterFilesxml(folder, exts):
    b=[]
    for fileName in os.listdir(folder):
        if os.path.isdir(folder + ‘/‘ + fileName):
            filterFiles(folder + ‘/‘ + fileName, exts)
        elif anyTrue(fileName.endswith, exts):
            sourceFile = codecs.open(folder + ‘/‘ + fileName,mode=‘r‘)
            xmlContentStr = sourceFile.read()
            xmlContentStr = xmlContentStr.decode(‘gb2312‘).encode(‘utf-8‘)
            xmlContentStr = xmlContentStr.replace(‘encoding="gb2312"‘,‘encoding="utf-8"‘)
            sourceXML = xml.dom.minidom.parseString(xmlContentStr)
            root = sourceXML.documentElement
            cnlist = root.getElementsByTagName(‘cnlink‘)
            for i in xrange(0,len(cnlist)):
                for ii in cnlist[i].firstChild.data.split(‘;‘):#有的值带有多条数据用;拼接，所以这里要分割一下
                    b.append(ii)
            enlist = root.getElementsByTagName(‘enlink‘)
            for i in xrange(0,len(enlist)):
                for ii in enlist[i].firstChild.data.split(‘;‘):
                    b.append(ii)
            tclist = root.getElementsByTagName(‘tclink‘)
            for i in xrange(0,len(tclist)):
                for ii in tclist[i].firstChild.data.split(‘;‘):
                    b.append(ii)
    b=list(set(b))#去重
    return b

def leakdattoleakxml(folder):
    exts_dat= [‘.dat‘]
    func=cdll.LoadLibrary(os.path.dirname(os.path.abspath(sys.argv[0]))+r‘\libleak_analyze.dll‘)
    get_xml_from_dat2  = func.get_xml_from_dat
    get_xml_from_dat2.argtypes = [c_wchar_p,c_wchar_p] # 参数类型
    get_xml_from_dat2.restype = c_int # 返回类型
    for fileName in os.listdir(folder):
        if os.path.isdir(folder + ‘/‘ + fileName):
            filterFiles(folder + ‘/‘ + fileName, exts_dat)
        elif anyTrue(fileName.endswith, exts_dat):     
            get_xml_from_dat2(folder + ‘/‘ + fileName , folder + ‘/‘ + fileName +‘.xml‘ )
    return True       
        
## ----------------每个线程执行一组url 
def perf1(folder): 
    exts_json = [‘.json‘]
    exts_xml = [‘.xml‘]
    # urls = filterFiles(folder, exts_json)##解析json
    urls = filterFilesxml(folder, exts_xml)##解析xml
    tsk = []
    for i in range(100):#并发数
        # print urls[i*100:i*100+100]#用列表是为了让每个线程跑多个url，而不用每个线程都只跑一个url
        t = threading.Thread(target=file_exists1,args=(urls[i*100:i*100+100],folder,))
        t.start()#启动线程
        tsk.append(t)    
    for tt in tsk:##用.join tsk表示所有线程执行完毕后，主线程才返回（从下面的end输出时间可以看出），它的缺点是大大降低线程执行效率，但是对于用multiprocessing.Process调用时比较有用，否则主线程返回后，改进程会被立即退出。如果是直接执行perf1()，建议不要用join
        tt.join()
    print ‘end‘    
    
if __name__ == ‘__main__‘: 
    ##把dat转成xml
    aa=r"E:\testtools\code\python\case\failleak\test"
    if leakdattoleakxml(aa):
        ##验证url
        perf1(aa)