How to speed up Python web crawler?

Question

def getProxiesLstFromDB():
    global g_proxies
    # insert to status page links table
    proxies_list = g_db["proxies_list"]
    proxies_list_records = proxies_list.find()
    print_to_log("get proxies list from db")
    for proxies_list_record in proxies_list_records:
        if proxies_list_record is not None:
            if len(proxies_list_record["user"]) == 0:
                proxies_list_record["user"] = "user"
            if len(proxies_list_record["password"]) == 0:
                proxies_list_record["password"] = "password"
            g_proxies.append(proxies_list_record)

def get_ProxyInfo(proxy_no):
    global g_proxies
    if len(g_proxies) == 0:
        return ""
    proxy_no = proxy_no % len(g_proxies)
    proxy_info = "%s:%s@%s:%s" % (g_proxies[proxy_no]["user"], g_proxies[proxy_no]  ["password"], g_proxies[proxy_no]["ProxyIP"], g_proxies[proxy_no]["ProxyPort"])
    return proxy_info

def readHtml_using_proxy(one_url, proxy_no):
    global g_proxies
    proxy = None
    page_text = ""
    try:
        proxy_info = "%s:%s@%s:%s" % (g_proxies[proxy_no]["user"], g_proxies[proxy_no]["password"], g_proxies[proxy_no]["ProxyIP"], g_proxies[proxy_no]["ProxyPort"])
        proxy = urllib2.ProxyHandler({"http": proxy_info,
                            "https":proxy_info})
    except:
        print_to_log("Proxy setting error. proxy_no: %d" % proxy_no)
        return page_text

    try:
        opener = urllib2.build_opener(proxy)
        req = urllib2.Request(one_url)
        r = opener.open(req)
        html = r.read()

        converter = html2text.HTML2Text()
        converter.ignore_links = True
        page_text = ""
        page_text = html.decode("utf8", "ignore")
        page_text = converter.handle(page_text)
    except:
        print_to_log("Unable to read or decode page html from %s" % one_url)

    return page_text

def scrape_one_url(db, one_url, domain_url, domain_url_idx, threadname):
    scrape_results = []
    threadnamelen = 13
    threadname = "(%s)" % threadname
    if len(threadname) < threadnamelen:
        idx = 0
        lenOfthreadname = len(threadname)
        while idx < threadnamelen - lenOfthreadname:
            idx += 1
            threadname = threadname + " "

    one_url = urljoin(domain_url, one_url)
    if "google.com" in one_url:
    #skip for google.com
        return
    if is_debug:
        print_to_log( "%s>>> processing domain_idx: %s, url: %s" % (threadname, domain_url_idx, one_url) )
    else:
        print_to_log( "%s>>> processing %s" % (threadname, one_url) )
    #get page content from url
    br = mechanize.Browser()
    br.set_handle_equiv(True)
    br.set_handle_redirect(True)
    br.set_handle_referer(True)
    br.set_handle_robots(False)

    try:
        response = br.open(one_url)
        page_type = response.info()["Content-Type"]
        if is_debug:
            print response.info()["Content-Type"]
        if "image" in page_type:
            return
        page_html = response.read()
    except:
        return


    converter = html2text.HTML2Text()
    converter.ignore_links = True

    try:
        page_text = page_html.decode("utf8", "ignore")
        page_text = converter.handle(page_text)
        page_text = re.sub("[^a-zA-Z0-9_!]+[ ,.\?!]", "", page_text)
    except:
        return

    #print page_text
    ###########################
    Site_Cate1 = ""
    Site_Cate2 = ""
    Site_Cate3 = ""
    Google_Page_Rank = "0"
    Twitter_share_cnt = ""
    Facebook_share_cnt = ""
    MAX_LIMIT_CNT_TO_INSERT = 100

    #1. Site Categorization
    xml = """<?xml version="1.0" encoding="utf-8" ?>
    <uclassify xmlns="http://api.uclassify.com/1/RequestSchema" version="1.01">
      <texts>
        <textBase64 id="TextId">%s</textBase64>
      </texts>
      <readCalls readApiKey="secret">
        <classify id="Clas" username="uClas" classifierName="Topic" textId="TextId"/>
      </readCalls>
    </uclassify>"""
    xml = xml % base64.b64encode(page_text.encode("utf8", "ignore"))
    headers = {'Content-Type': 'text/xml; charset=utf-8'} # set what your server accepts
    response = requests.post('http://api.uclas.com', data=xml, headers=headers).text
    res_xml = BeautifulSoup(response)
    Cate_lst = {}
    try:
        for one_class in res_xml.findAll("class"):
        Cate_lst[one_class.attrs["classname"]] = float(one_class.attrs["p"])
    except:
        pass

    idx = 1
    for key, value in sorted(Cate_lst.iteritems(), key=lambda (k,v): (v,k),  reverse=True):
        if idx == 1:
            Site_Cate1 = key
        elif idx == 2:
            Site_Cate2 = key
        elif idx == 3:
            Site_Cate3 = key
            break
        idx += 1

    global g_proxy_counter
    g_queueLock.acquire()
    g_proxy_counter += 1
    g_queueLock.release()

    proxy_info = get_ProxyInfo(g_proxy_counter)
    try:
        if len(proxy_info) > 0:
            Google_Page_Rank = rank_provider.GooglePageRank().get_rank_using_proxy(one_url, proxy_info)
            if Google_Page_Rank is None:
                Google_Page_Rank = "0"
            else:
                Google_Page_Rank = str(Google_Page_Rank)
    except:
        Google_Page_Rank = "0"
    if is_debug:
        print_to_log( "%sGoogle Page Rank: %s" % (threadname, Google_Page_Rank) )
    #8. Social Share Counts
    one_url_tw = "http://cdn.api.twitter.com/1/urls/count.json?url=%s" % one_url

    try:
        response_tw = br.open(one_url_tw)
        page_html_tw = response_tw.read()
        j = json.loads(page_html_tw)
        Twitter_share_cnt = j['count']
    except:
        pass
    one_url_fb = "http://graph.facebook.com/?id=%s" % one_url
    try:
        response_fb = br.open(one_url_fb)
        page_html_fb = response_fb.read()
        j = json.loads(page_html_fb)
        Facebook_share_cnt = j['shares']
    except:
        pass

    #5. Total Backlinks
    t_soup = BeautifulSoup(page_html)
    t_as = t_soup.findAll('a')
    Total_Incoming_links = 0
    Total_Outgoing_links = 0
    links_idx = 0
    records = []
    status_records = []
    for t_a in t_as:
        if not t_a.has_attr('href') or (not t_a['href'].startswith("http") and not t_a['href'].startswith("/")):
            continue
        #/#comment, /#readmore, /#respond
        if ("/#comment" in t_a['href']) or ("/#readmore" in t_a['href']) or ("/#respond" in t_a['href']):
            continue
        domain_url_www = domain_url.replace('https://','https://www.').replace('http://','http://www.')
        if t_a['href'].startswith(domain_url):
            t_a['href'] = t_a['href'][len(domain_url):]
        elif t_a['href'].startswith(domain_url_www):
            t_a['href'] = t_a['href'][len(domain_url_www):]
        InOrOutLink = ""
        if t_a['href'].startswith("http"):
            Total_Outgoing_links += 1
            InOrOutLink = "Outgoing"
        else:
            Total_Incoming_links += 1
            InOrOutLink = "Incoming"
            #################################
            # insert to status page links table
            status_page_links = db["status_page_links"]
            status_page_links_record = status_page_links.find_one({"DomainURL": domain_url, "DomainURLIDX": domain_url_idx, "Link": t_a['href']})
            if status_page_links_record is None:
                status_record = {
                    "DomainURL": domain_url,
                    "DomainURLIDX": domain_url_idx,
                    "Link": t_a['href'],
                    "Status": 0, #0: not processed, 1: processed
                    "date": datetime.datetime.utcnow()}
                status_page_links_id = status_page_links.insert(status_record)
                scrape_results.append(status_record)

        links_idx += 1
        ###################################
        # insert to db
        record = {
            "DomainURL": domain_url,
            "DomainURLIDX": domain_url_idx,
            "PageURL": one_url,
            "Link": t_a['href'],
            "InOrOutLink": InOrOutLink,
            "date": datetime.datetime.utcnow()}
        records.append(record)

        if links_idx % MAX_LIMIT_CNT_TO_INSERT == 0:
            page_link_info = db["page_link_info"]
            page_link_info_ids = page_link_info.insert(records)
            records = []
            if is_debug:
                print_to_log( "%sinserted %s links" % (threadname, links_idx) )
        ####################################
        #calculate total count of incoming / outcoming links

    if links_idx % MAX_LIMIT_CNT_TO_INSERT > 0:
        page_link_info = db["page_link_info"]
        page_link_info_ids = page_link_info.insert(records)
        records = []

    #2. Keyword Relevance, 3. Key word Sentiment Value
    extractor = extract.TermExtractor()
    extractor.filter = extract.DefaultFilter(noLimitStrength=2)
    kwds = sorted( extractor(page_text) )
    total_cntOfWds = len([word for word in page_text.split() if word.isalnum()])
    KeywordsCnt = 0
    xml_sentiment_templ = """<?xml version="1.0" encoding="utf-8" ?>
<uclassify xmlns="http://api.uclassify.com/1/RequestSchema" version="1.01">
<texts>
 <textBase64 id="tweet1">%s</textBase64>
</texts>
<readCalls readApiKey="secret">
<classifyKeywords id="ClassifyKeywords" username="uClassify" classifierName="Sentiment"     textId="tweet1"/>
</readCalls>
</uclassify>"""
    records = []
    for kwd in kwds:
        #break # for test
        (one_word, occurences, cntOfWd) = kwd
        if re.search('[a-zA-Z]+',one_word) == None:
            pass
        elif len(one_word) < 3:
            pass
        elif cntOfWd > 3:
            pass
        else:
            Keyword_Density = float(occurences) * cntOfWd / total_cntOfWds
            Keyword_Density = "%.6f" % round(float(Keyword_Density),6)
            try:
                one_word = one_word.decode("utf8", "ignore")
                xml_sentiment = xml_sentiment_templ % base64.b64encode(one_word.encode("utf8", "ignore"))
            except:
                continue

            KS_positive = 0
            KS_negative = 0
            KS_type = "neutral"
            #check if kw exists in db already

            KeywordsCnt += 1

            ###################################
            # insert to db
            record = {
                "DomainURL": domain_url,
                "DomainURLIDX": domain_url_idx,
                "PageURL": one_url,
                "Keyword": one_word,
                "KWSentiment": KS_type,
                "KWSPositive": KS_positive,
                "KWSNegative": KS_negative,
                "KWDensity": Keyword_Density,
                "KWOccurences": occurences,
                "KWCntOfWords": cntOfWd,
                "date": datetime.datetime.utcnow()}
            records.append(record)

            if KeywordsCnt % MAX_LIMIT_CNT_TO_INSERT == 0:
                page_kw_info = db["page_kw_info"]
                page_kw_info_ids = page_kw_info.insert(records)
                records = []
            ####################################

     if KeywordsCnt % MAX_LIMIT_CNT_TO_INSERT > 0:
        page_kw_info = db["page_kw_info"]
        page_kw_info_ids = page_kw_info.insert(records)
        records = []

    ####################################

    #6. Kewyword Density
    # already got

    #7. Domain Age

    #8. Calculate Points
    Points = 0
    try:
        int_Google_Page_Rank = int(Google_Page_Rank)
    except:
        int_Google_Page_Rank = 0

    try:
        int_Total_Incoming_links = int(Total_Incoming_links)
    except:
        int_Total_Incoming_links = 0

    try:
        int_Total_Outgoing_links = int(Total_Outgoing_links)
    except:
        int_Total_Outgoing_links = 0

    try:
        int_Twitter_share_cnt = int(Twitter_share_cnt)
    except:
        int_Twitter_share_cnt = 0

    try:
        int_Facebook_share_cnt = int(Facebook_share_cnt)
    except:
        int_Facebook_share_cnt = 0

    try:
        Points = int_Google_Page_Rank*10 + int_Total_Incoming_links/10 - int_Total_Outgoing_links/5 + ( int_Twitter_share_cnt + int_Facebook_share_cnt )/15
    except:
        Points = 0
    ###################################
    # insert to db
    record = {
        "DomainURL": domain_url,
        "DomainURLIDX": domain_url_idx,
        "PageURL": one_url,
        "SiteCate1": Site_Cate1,
        "SiteCate2": Site_Cate2,
        "SiteCate3": Site_Cate3,
        "GooglePageRank": Google_Page_Rank,
        "FacebookShareCnt": Facebook_share_cnt,
        "TwitterShareCnt": Twitter_share_cnt,
        "TotalBacklinks": Total_Incoming_links + Total_Outgoing_links,
        "TotalIncomingLinksCnt": Total_Incoming_links,
        "TotalOutgoingLinksCnt": Total_Outgoing_links,
        "TotalWordsCnt": total_cntOfWds,
        "TotalKeywordsCnt": KeywordsCnt,
        "Points": Points,
        "date": datetime.datetime.utcnow()}
    page_main_info = db["page_main_info"]
    update_status = page_main_info.update({"PageURL":one_url}, record, upsert=True)
    return scrape_results

def scrape_one_domain(domain_url, domain_url_idx, threadname):
    global PAGES_CRAWLING_THREADS
    one_url = "/"
    client = MongoClient('secret', 2422)
    db = client['site_analysis']
    # insert to status page links table
    status_page_links = db["status_page_links"]
    status_page_links_record = status_page_links.find_one({"DomainURL": domain_url, "DomainURLIDX": domain_url_idx, "Link": one_url})
    if status_page_links_record is None:
        status_record = {
            "DomainURL": domain_url,
            "DomainURLIDX": domain_url_idx,
            "Link": one_url,
            "Status": 0, #0: not processed, 1: processed
            "date": datetime.datetime.utcnow()}
        status_page_links_id = status_page_links.insert(status_record)


 #process not processed links in separate threads
def process_link_worker(queue, db, domain_url, domain_url_idx, threadname, threadsubname, threads_signals):
    global PAGES_CRAWLING_THREADS
    while True:
        try:
            status_page_links_record = queue.get_nowait()
        except Queue.Empty:
            threads_signals.add(threadsubname)
            if len(threads_signals) >= PAGES_CRAWLING_THREADS * 10: # if no job left
                return
            else:
                time.sleep(1)
                continue
        threads_signals.discard(threadsubname)
        one_url = status_page_links_record["Link"]
        scrape_results = scrape_one_url(db, one_url, domain_url, domain_url_idx, threadname)
        for scrape_result in scrape_results:
            queue.put(scrape_result)
        db["status_page_links"].update({'_id': ObjectId(status_page_links_record["_id"])},
                                    {'$set': {"Status": 1, "date": datetime.datetime.utcnow()}})

status_page_links = db["status_page_links"]
status_page_links_records = list(status_page_links.find({"DomainURL": domain_url, "DomainURLIDX": domain_url_idx, "Status": 0}))
if status_page_links_records:
    links_queue = Queue.Queue()
    for status_page_links_record in status_page_links_records:
        links_queue.put(status_page_links_record)
    link_processing_threads = []
    threads_signals = set()
    for i in range(PAGES_CRAWLING_THREADS): #start 2 threads
        thread = threading.Thread(target=process_link_worker, args=(links_queue,db,domain_url,domain_url_idx,
                                                                    threadname,'%s_%s' % (threadname, i+1),
                                                                    threads_signals))
        thread.start()
        link_processing_threads.append(thread)
    #wait for threads return
    for thread in link_processing_threads:
        thread.join()


exitFlag = 0

class myThread (threading.Thread):
    def __init__(self, threadID, name, q):
        threading.Thread.__init__(self)
        self.threadID = threadID
        self.name = name
        self.q = q
    def run(self):
        global g_queueLock
        global g_workQueue
        global g_workingEntry
        global g_db
        print_to_log( "Starting " + self.name )
        while not exitFlag:
            g_queueLock.acquire()
            if not g_workQueue.empty():
                data = g_workQueue.get()
                domain_url = data[0]
                domain_url_idx = data[1]
                g_workingEntry.append(domain_url)
                g_queueLock.release()
                scrape_one_domain(domain_url, domain_url_idx, self.name)
                #updated domain status in db
                g_queueLock.acquire()
                g_workingEntry.remove(domain_url)
                status_domain = g_db["status_domain"]
                status_domain.update({"DomainURL": domain_url.strip(), "DomainURLIDX": domain_url_idx,"Status": 0},
                {'$set': {"Status": 1, "date": datetime.datetime.utcnow()}})
                g_queueLock.release()
            else:
                g_queueLock.release()
            time.sleep(1)

        print_to_log( "Exiting " + self.name )


def main():
    global g_threads
    global g_workQueue
    global g_db
    global g_workingEntry
    global g_queueLock


    if len(sys.argv) < 2:
        print_to_log("================Crawler newly started==============")
        print_to_log("you did not give any arguments.")
        threadCnt = 10
    else:
        print_to_log("================Crawler newly started==============")
        threadCnt = int(sys.argv[1])

    getProxiesLstFromDB()
    print_to_log("Threads Count: %d" % threadCnt)

    # Create new threads
    threadID = 1
    while threadID <= threadCnt:
        threadName = "Thread %d" % threadID
        thread = myThread(threadID, threadName, g_workQueue)
        thread.start()
        g_threads.append(thread)
        threadID += 1

    is_first = True
    while True:
        # insert to status page links table
        status_domain = g_db["status_domain"]
        status_domain_records_cnt = status_domain.find({"Status": 0}).count()
        if (is_first and status_domain_records_cnt == 0) or (is_first == False):
            #read domains
            domains_to_crawl = g_db["domains_to_crawl"]
            domains_to_crawl_records = domains_to_crawl.find()
            for domains_to_crawl_record in domains_to_crawl_records:
                max_status_domain_record = status_domain.find_one(sort=[("DomainURLIDX", -1)])
                max_DomainURLIDX = 0
                if max_status_domain_record is not None:
                    max_DomainURLIDX = max_status_domain_record["DomainURLIDX"]

                status_domain_record = {
                    "DomainURL": domains_to_crawl_record["DomainURL"],
                    "DomainURLIDX": max_DomainURLIDX+1,
                    "Status": 0, #0: not processed, 1: processed
                    "date": datetime.datetime.utcnow()}
                try:
                    index = g_workingEntry.index(domains_to_crawl_record["DomainURL"])
                except:
                    update_status = status_domain.update({"DomainURL": domains_to_crawl_record["DomainURL"].strip()}, status_domain_record, upsert=True)
                    print_to_log("domains_to_crawl table content: %s" % domains_to_crawl_record["DomainURL"] )

        status_domain_records = status_domain.find({"Status": 0}).sort("DomainURLIDX", 1)
        # Fill the queue
        g_queueLock.acquire()
        while not g_workQueue.empty():
            data = g_workQueue.get()
        for record in status_domain_records:
            try:
                index = g_workingEntry.index(record["DomainURL"])
            except:
                #not exist the url in working list
                g_workQueue.put([record["DomainURL"], record["DomainURLIDX"]])
        g_queueLock.release()

        # Wait for queue to empty
        while not g_workQueue.empty():
            pass
        is_first = False
        time.sleep(1)
    # Notify threads it's time to exit
    exitFlag = 1

    # Wait for all threads to complete
    for t in g_threads:
        t.join()
    print_to_log( "Exiting Main Thread" )

Answer 1

This is just one bit that jumped out at me:

    page_text = page_html.decode("utf8", "ignore")
    page_text = converter.handle(page_text)
    page_text = re.sub("[^a-zA-Z0-9_!]+[ ,.\?!]", "", page_text)

It seems you are converting as utf8, ignoring errors, and then forcing it into ascii with a regex. Why not just decode as ascii and ignore or replace the errors.

There are certainly many other optimizations in a chunk of code that size. You really are doing alot in that chunk of code and there are so many ways to improve on it.

Another observation is the page_html.split() to glean the words out of the document. You have used BeautifulSoup else where in the doc why not use that to get the words. Preferably all in one global parse pass.

Further, you then turn around and submit those words each to an API. It doesn't look like there is any thought that the same word might return the same result each time and as such the results could be stored in some other way so as to avoid needing to request them for each word on each page, which surely will lead to many many redundant queries against the classification.

There is really too much to cover in a single response here.