mrq/examples/simple_crawler/crawler.py at 3f23427e3f473506bfd5bcd828c3b7cf4bd91aed · pricingassistant/mrq · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
import requests
import lxml.html
import datetime
import re
import urlparse
from mrq.context import connections, log
from mrq.job import queue_job
from mrq.task import Task
from mrq.queue import Queue


class Fetch(Task):

    def run(self, params):

        collection = connections.mongodb_jobs.simple_crawler_urls

        response = requests.get(params["url"])

        if response.status_code != 200:
            log.warning("Got status %s on page %s (Queued from %s)" % (
                response.status_code, response.url, params.get("from")
            ))
            return False

        # Store redirects
        if response.url != params["url"]:
            collection.update({"_id": params["url"]}, {"$set": {
                "redirected_to": response.url,
                "fetched_date": datetime.datetime.now()
            }})

        document = lxml.html.fromstring(response.content)

        document.make_links_absolute(response.url)

        queued_count = 0

        document_domain = urlparse.urlparse(response.url).netloc

        for (element, attribute, link, pos) in document.iterlinks():

            link = re.sub("#.*", "", link or "")

            if not link:
                continue

            domain = urlparse.urlparse(link).netloc

            # Don't follow external links for this example
            if domain != document_domain:
                continue

            # We don't want to re-queue URLs twice. If we try to insert a duplicate,
            # pymongo will throw an error
            try:
                collection.insert({"_id": link})
            except:
                continue

            queue_job("crawler.Fetch", {
                "url": link,
                "from": params["url"]
            }, queue="crawl")
            queued_count += 1

        stored_data = {
            "_id": response.url,
            "queued_urls": queued_count,
            "html_length": len(response.content),
            "fetched_date": datetime.datetime.now()
        }

        collection.update(
            {"_id": response.url},
            stored_data,
            upsert=True
        )

        return True


class Report(Task):

    def run(self, params):

        collection = connections.mongodb_jobs.simple_crawler_urls

        print
        print "Crawl stats"
        print "==========="
        print "URLs queued: %s" % collection.find().count()
        print "URLs successfully crawled: %s" % collection.find({"fetched_date": {"$exists": True}}).count()
        print "URLs redirected: %s" % collection.find({"redirected_to": {"$exists": True}}).count()
        print "Bytes fetched: %s" % (list(collection.aggregate(
            {"$group": {"_id": None, "sum": {"$sum": "$html_length"}}}
        )) or [{}])[0].get("sum", 0)
        print


class Reset(Task):

    def run(self, params):

        collection = connections.mongodb_jobs.simple_crawler_urls

        collection.remove({})

        Queue("crawl").empty()