commited initial scrapy code, taken from the old repo at r31560

--HG--
extra : convert_revision : svn%3Ab85faa78-f9eb-468e-a121-7cced6da292c%4025
This commit is contained in:
Pablo Hoffman 2008-06-28 23:37:28 +00:00
parent 37ca8701eb
commit 83dcf8aff9
215 changed files with 18021 additions and 0 deletions

31
scrapy/trunk/INSTALL Normal file
View File

@ -0,0 +1,31 @@
You need:
* Python 2.5
* Twisted
* libxml2 python bindings. version 2.6.28 or above is highly recommended.
* pyopenssl for HTTPS support
* for win32: http://webcleaner.sourceforge.net/pyOpenSSL-0.6.win32-py2.5.exe
* spidermonkey (optional) - required for JSParser
The python interpreter should have sys.getdefaultencoding() == 'utf-8'
If this is not the case, you can add the following lines to a file
[PYTHON_INSTALL]/site-packages/sitecustomize.py:
import sys
sys.setdefaultencoding('utf-8')
INSTALLING LIBRARIES
--------------------
The procedure for installing the required third party libraries (twisted,
libxml2 and pyopenssl) depends on the platform and OS you use.
Visit the project homepages for more information about it:
* Twisted: http://twistedmatrix.com/
* libxml2: http://xmlsoft.org/
* releases available here: ftp://xmlsoft.org/libxml2/
* pyopenssl: http://pyopenssl.sourceforge.net/
In Debian/Ubuntu Linux it's as easy as:
apt-get install python-twisted python-libxml2 python-pyopenssl

3
scrapy/trunk/README Normal file
View File

@ -0,0 +1,3 @@
This is Scrapy, an opensource screen scraping framework written in Python.
For more visit the project home page at http://scrapy.org

View File

@ -0,0 +1,72 @@
DROP TABLE IF EXISTS `url_history`;
DROP TABLE IF EXISTS `version`;
DROP TABLE IF EXISTS `url_status`;
DROP TABLE IF EXISTS `ticket`;
DROP TABLE IF EXISTS `domain_stats`;
DROP TABLE IF EXISTS `domain_stats_history`;
DROP TABLE IF EXISTS `domain_data_history`;
CREATE TABLE `ticket` (
`guid` char(40) NOT NULL,
`domain` varchar(255) default NULL,
`url` varchar(2048) default NULL,
`url_hash` char(40) default NULL, -- so we can join to url_status
PRIMARY KEY (`guid`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
CREATE TABLE `version` (
`id` bigint(20) NOT NULL auto_increment,
`guid` char(40) NOT NULL,
`version` char(40) NOT NULL,
`seen` datetime NOT NULL,
PRIMARY KEY (`id`),
FOREIGN KEY (`guid`) REFERENCES ticket(guid) ON UPDATE CASCADE ON DELETE CASCADE,
UNIQUE KEY (`version`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
CREATE TABLE `url_status` (
-- see http://support.microsoft.com/kb/q208427/ for explanation of 2048
`url_hash` char(40) NOT NULL, -- for faster searches
`url` varchar(2048) NOT NULL,
`parent_hash` char(40) default NULL, -- the url that was followed to this one - for reporting
`last_version` char(40) default NULL, -- can be null if it generated an error the last time is was checked
`last_checked` datetime NOT NULL,
PRIMARY KEY (`url_hash`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
CREATE TABLE `url_history` (
`url_hash` char(40) NOT NULL,
`version` char(40) NOT NULL,
`postdata_hash` char(40) default NULL,
`created` datetime NOT NULL,
PRIMARY KEY (`version`),
FOREIGN KEY (`url_hash`) REFERENCES url_status(url_hash) ON UPDATE CASCADE ON DELETE CASCADE
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
CREATE TABLE `domain_stats` (
`key1` varchar(128) NOT NULL,
`key2` varchar(128) NOT NULL,
`value` text,
PRIMARY KEY `key1_key2` (`key1`, `key2`),
KEY `key1` (`key1`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
CREATE TABLE `domain_stats_history` (
`id` bigint(20) NOT NULL auto_increment,
`key1` varchar(128) NOT NULL,
`key2` varchar(128) NOT NULL,
`value` varchar(2048) NOT NULL,
`stored` datetime NOT NULL,
PRIMARY KEY (`id`),
KEY `key1_key2` (`key1`, `key2`),
KEY `key1` (`key1`),
KEY `stored` (`stored`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
CREATE TABLE `domain_data_history` (
`domain` varchar(255) NOT NULL,
`stored` datetime NOT NULL,
`data` text,
KEY `domain_stored` (`domain`, `stored`),
KEY `domain` (`domain`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

View File

@ -0,0 +1,18 @@
"""
Scrapy - a screen scraping framework written in Python
"""
__version__ = "0.1.0"
import sys, os
if sys.version_info < (2,5):
print "Scrapy %s requires Python 2.5 or above" % __version__
sys.exit(1)
# add external python libraries bundled into scrapy
sys.path.insert(0, os.path.join(os.path.abspath(os.path.dirname(__file__)), "lib"))
# monkey patches to fix external library issues
from scrapy.patches import monkeypatches
monkeypatches.apply_patches()

View File

@ -0,0 +1 @@
from scrapy.command.models import ScrapyCommand

View File

@ -0,0 +1,154 @@
from __future__ import with_statement
import sys
import os
import optparse
import scrapy
from scrapy.core import log
from scrapy.spider import spiders
from scrapy.conf import settings
def find_commands(dir):
try:
return [f[:-3] for f in os.listdir(dir) if not f.startswith('_') and f.endswith('.py')]
except OSError:
return []
def builtin_commands_dict():
d = {}
scrapy_dir = scrapy.__path__[0]
commands_dir = os.path.join(scrapy_dir, 'command', 'commands')
for cmdname in find_commands(commands_dir):
modname = 'scrapy.command.commands.%s' % cmdname
command = getattr(__import__(modname, {}, {}, [cmdname]), 'Command', None)
if callable(command):
d[cmdname] = command()
else:
print 'WARNING: Builtin command module %s exists but Command class not found' % modname
return d
def custom_commands_dict():
d = {}
cmdsmod = settings['COMMANDS_MODULE']
if cmdsmod:
mod = __import__(cmdsmod, {}, {}, [''])
for cmdname in find_commands(mod.__path__[0]):
modname = '%s.%s' % (cmdsmod, cmdname)
command = getattr(__import__(modname, {}, {}, [cmdname]), 'Command', None)
if callable(command):
d[cmdname] = command()
else:
print 'WARNING: Custom command module %s exists but Command class not found' % modname
return d
def getcmdname():
for arg in sys.argv[1:]:
if not arg.startswith('-'):
return arg
def usage():
s = "usage: %s <subcommand> [options] [args]\n" % sys.argv[0]
s += " %s <subcommand> -h\n\n" % sys.argv[0]
s += "Built-in subcommands:\n"
builtin_cmds = builtin_commands_dict()
custom_cmds = custom_commands_dict()
filtered_builtin_cmds = [(name, cls) for name, cls in builtin_cmds.iteritems() if name not in custom_cmds]
for cmdname, cmdclass in filtered_builtin_cmds:
s += " %s %s\n" % (cmdname, cmdclass.syntax())
s += " %s\n" % cmdclass.short_desc()
if custom_cmds:
s += "\n"
s += "Custom (or overloaded) subcommands:\n"
for cmdname, cmdclass in custom_cmds.iteritems():
s += " %s %s\n" % (cmdname, cmdclass.syntax())
s += " %s\n" % cmdclass.short_desc()
return s
def update_defaults(defaults, module):
settingsdict = vars(module)
for k, v in settingsdict.iteritems():
if not k.startswith("_"):
defaults[k] = v
def command_settings(cmdname):
try:
module = __import__('%s.%s' % ('scrapy.conf.commands', cmdname), {}, {}, [''])
update_defaults(settings.defaults, module)
except ImportError:
pass
basepath = settings['COMMANDS_SETTINGS_MODULE']
if basepath:
try:
module = __import__('%s.%s' % (basepath, cmdname), {}, {}, [''])
update_defaults(settings.defaults, module)
except ImportError:
pass
# This dict holds information about the executed command for later use
command_executed = {}
def execute():
spiders.load()
cmds = builtin_commands_dict()
cmds.update(custom_commands_dict())
cmdname = getcmdname()
command_settings(cmdname)
if not cmdname:
print "Scrapy %s\n" % scrapy.__version__
print usage()
sys.exit()
parser = optparse.OptionParser()
if cmdname in cmds:
cmd = cmds[cmdname]
cmd.add_options(parser)
parser.usage = "%%prog %s %s" % (cmdname, cmd.syntax())
parser.description = cmd.long_desc()
else:
print "Scrapy %s\n" % scrapy.__version__
print "Unknown command: %s\n" % cmdname
print 'Type "%s -h" for help' % sys.argv[0]
sys.exit()
(opts, args) = parser.parse_args()
del args[0] # args[0] is cmdname
# storing command executed info for later reference
command_executed['name'] = cmdname
command_executed['class'] = cmd
command_executed['args'] = args[:]
command_executed['opts'] = opts.__dict__.copy()
cmd.process_options(args, opts)
log.start() # start logging
if opts.profile:
log.msg("Profiling enabled. Analyze later with: python -m pstats %s" % opts.profile)
import cProfile
loc = locals()
p = cProfile.Profile()
p.runctx('ret = cmd.run(args, opts)', globals(), loc)
p.dump_stats(opts.profile)
try:
import lsprofcalltree
fn = opts.profile + ".cachegrind"
k = lsprofcalltree.KCacheGrind(p)
with open(fn, 'w') as f:
k.output(f)
except ImportError:
pass
ret = loc['ret']
else:
ret = cmd.run(args, opts)
if ret is False:
parser.print_help()

View File

@ -0,0 +1,67 @@
"""
Commmands for controlling Scrapy
"""
from scrapy.conf import settings
class ScrapyCommand(object):
def syntax(self):
"""
Command syntax (preferably one-line). Do not include command name.
"""
return ""
def short_desc(self):
"""
A short description of the command
"""
return ""
def long_desc(self):
"""
A long description of the command. Return short description when not
available. It cannot contain newlines, since contents will be formatted
by optparser which removes newlines and wraps text.
"""
return self.short_desc()
def help(self):
"""
An extensive help for the command. It will be shown when using the
"help" command. It can contain newlines, since not post-formatting will
be applied to its contents.
"""
return self.long_desc()
def add_options(self, parser):
"""
Populate option parse with options available for this command
"""
parser.add_option("-f", "--logfile", dest="logfile", help="logfile to use. if omitted stderr will be used", metavar="FILE")
parser.add_option("-o", "--loglevel", dest="loglevel", default=None, help="log level")
parser.add_option("--spider", dest="spider", default=None, help="default spider (domain) to use if no spider is found")
parser.add_option("--nolog", dest="nolog", action="store_true", help="disable all log messages")
parser.add_option("--profile", dest="profile", default=None, help="write profiling stats in FILE, to analyze later with: python -m pstats FILE", metavar="FILE")
def process_options(self, args, opts):
if opts.logfile:
settings.overrides['LOG_ENABLED'] = True
settings.overrides['LOGFILE'] = opts.logfile
if opts.loglevel:
settings.overrides['LOG_ENABLED'] = True
settings.overrides['LOGLEVEL'] = opts.loglevel
if opts.nolog:
settings.overrides['LOG_ENABLED'] = False
if opts.spider:
from scrapy.spider import spiders
spiders.default_domain = opts.spider
def run(self, args, opts):
"""
Entry point for running commands
"""
raise NotImplementedError

View File

@ -0,0 +1,38 @@
from scrapy.command import ScrapyCommand
from scrapy.core.manager import scrapymanager
from scrapy.replay import Replay
from scrapy.conf import settings
class Command(ScrapyCommand):
def syntax(self):
return "[options] [domain|url] ..."
def short_desc(self):
return "Run the web scraping engine from the command line"
def add_options(self, parser):
ScrapyCommand.add_options(self, parser)
parser.add_option("--restrict", dest="restrict", action="store_true", help="restrict crawling only to the given urls")
parser.add_option("--record", dest="record", help="use FILE for recording session (see replay command)", metavar="FILE")
parser.add_option("--record-dir", dest="recorddir", help="use DIR for recording (instead of file)", metavar="DIR")
def process_options(self, args, opts):
ScrapyCommand.process_options(self, args, opts)
if opts.nopipeline:
settings.overrides['ITEM_PIPELINES'] = []
if opts.nocache:
settings.overrides['CACHE2_DIR'] = None
if opts.restrict:
settings.overrides['RESTRICT_TO_URLS'] = args
if opts.record or opts.recorddir:
# self.replay is used for preventing Replay signals handler from
# disconnecting since pydispatcher uses weak references
self.replay = Replay(opts.record or opts.recorddir, mode='record', usedir=bool(opts.recorddir))
self.replay.record(args=args, opts=opts.__dict__)
def run(self, args, opts):
scrapymanager.runonce(*args, **opts.__dict__)

View File

@ -0,0 +1,28 @@
import sys, pprint
from scrapy.command import ScrapyCommand
from scrapy.fetcher import fetch
class Command(ScrapyCommand):
def syntax(self):
return "[options] <url>"
def short_desc(self):
return "Download a URL using the Scrapy downloader"
def add_options(self, parser):
ScrapyCommand.add_options(self, parser)
parser.add_option("-s", "--source", dest="source", action="store_true", help="output HTTP body only")
parser.add_option("--headers", dest="headers", action="store_true", help="output HTTP headers only")
def run(self, args, opts):
if not args:
print "A URL is required"
return
responses = fetch(args)
if responses:
if opts.headers:
pprint.pprint(responses[0].headers)
else:
sys.stdout.write(str(responses[0].body))

View File

@ -0,0 +1,52 @@
import os
from scrapy.spider import spiders
from scrapy.command import ScrapyCommand
from scrapy.conf import settings
class Command(ScrapyCommand):
def syntax(self):
return "<spider_name> <spider_domain_name>"
def short_desc(self):
return "Generate new spider based on predefined template"
def run(self, args, opts):
if len(args) != 2:
return False
name = args[0]
site = args[1]
spiders_dict = spiders.asdict()
if not name in spiders_dict.keys():
self._genspider(name, site)
else:
print "Spider '%s' exist" % name
def _genspider(self, name, site):
""" Generate spider """
tvars = {
'name': name,
'site': site,
'classname': '%sSpider' % ''.join([s.capitalize() for s in name.split('-')])
}
spiders_module = __import__(settings['NEWSPIDER_MODULE'], {}, {}, [''])
spidersdir = os.path.abspath(os.path.dirname(spiders_module.__file__))
self._genfiles('spider.tmpl', '%s/%s.py' % (spidersdir, name), tvars)
def _genfiles(self, template_name, source_name, tvars):
""" Generate source from template, substitute variables """
template_file = os.path.join(settings['TEMPLATES_DIR'], template_name)
tmpl = open(template_file)
clines = []
for l in tmpl.readlines():
for key, val in tvars.items():
l = l.replace('@%s@' % key, val)
clines.append(l)
tmpl.close()
source = ''.join(clines)
if not os.path.exists(source_name):
sfile = open(source_name, "w")
sfile.write(source)
sfile.close()

View File

@ -0,0 +1,25 @@
from scrapy.command import ScrapyCommand
from scrapy.fetcher import fetch
from scrapy.spider import spiders
from scrapy.item import ScrapedItem
def get_item_attr(pagedata, attr="guid"):
spider = spiders.fromurl(pagedata.url)
items = spider.parse(pagedata)
attrs = [getattr(i, attr) for i in items if isinstance(i, ScrapedItem)]
return attrs
def get_attr(url, attr="guid"):
pagedatas = fetch([url])
if pagedatas:
return get_item_attr(pagedatas[0], attr)
class Command(ScrapyCommand):
def syntax(self):
return "<url> <attribute>"
def short_desc(self):
return "Print an attribute from the item scraped in the given URL"
def run(self, args, opts):
print get_attr(args[0], args[1])

View File

@ -0,0 +1,26 @@
from scrapy.command import ScrapyCommand, cmdline
class Command(ScrapyCommand):
def syntax(self):
return "<command>"
def short_desc(self):
return "Provides extended help for the given command"
def run(self, args, opts):
if not args:
return False
commands = cmdline.builtin_commands_dict()
commands.update(cmdline.custom_commands_dict())
cmdname = args[0]
if cmdname in commands:
cmd = commands[cmdname]
help = getattr(cmd, 'help', None) or getattr(cmd, 'long_desc', None)
print "%s: %s" % (cmdname, cmd.short_desc())
print "usage: %s %s" % (cmdname, cmd.syntax())
print
print help()
else:
print "Unknown command: %s" % cmdname

View File

@ -0,0 +1,16 @@
from scrapy.command import ScrapyCommand
from scrapy.spider import spiders
class Command(ScrapyCommand):
def syntax(self):
return ""
def short_desc(self):
return "List available spiders (both enabled and disabled)"
def run(self, args, opts):
spiders_dict = spiders.asdict()
for n, p in spiders_dict.items():
disabled = "disabled" if getattr(p, 'disabled', False) else "enabled"
print "%-30s %-30s %s" % (n, p.__class__.__name__, disabled)
print "Total spiders: %d" % len(spiders_dict)

View File

@ -0,0 +1,46 @@
import re
from scrapy.command import ScrapyCommand
log_crawled_re = re.compile(r'^(.*?) \[\w+/(.*?)\].*Crawled <(.*?)> from <(.*?)>$')
log_scraped_re = re.compile('^(.*?) \[\w+/(.*?)\].*Scraped (.*) in <(.*?)>$')
log_opendomain_re = re.compile('^(.*?) \[\w+.*? Started scraping (.*)$')
log_closedomain_re = re.compile('^(.*?) \[\w+.*? Finished scraping (.*)$')
log_debug_re = re.compile(r'\[\w+/(.*?)\] DEBUG')
log_info_re = re.compile(r'\[\w+/(.*?)\] INFO')
log_warning_re = re.compile(r'\[\w+/(.*?)\] WARNING')
log_error_re = re.compile(r'\[\w+/(.*?)\] ERROR')
class Command(ScrapyCommand):
def syntax(self):
return "[options] <logfile>"
def short_desc(self):
return "Several tools to perform scrapy log analysis"
def add_options(self, parser):
ScrapyCommand.add_options(self, parser)
parser.add_option("--sitemap", dest="sitemap", action="store_true", help="show sitemap based on given log file")
def sitemap(self, logfile):
from scrapy.utils.datatypes import Sitemap
sm = Sitemap()
for l in open(logfile).readlines():
m = log_crawled_re.search(l.strip())
if m:
sm.add_node(m.group(3), m.group(4))
m = log_scraped_re.search(l.strip())
if m:
sm.add_item(m.group(4), m.group(3))
print sm.to_string()
def run(self, args, opts):
if not args:
print "A log file is required"
return
if opts.sitemap:
self.sitemap(args[0])
else:
print "No analysis method specified"

View File

@ -0,0 +1,46 @@
from scrapy.command import ScrapyCommand
from scrapy.fetcher import fetch
from scrapy.http import Request
from scrapy.item import ScrapedItem
from scrapy.spider import spiders
from scrapy.utils import display
class Command(ScrapyCommand):
def syntax(self):
return "[options] <url>"
def short_desc(self):
return "Parse the URL and print their results"
def add_options(self, parser):
ScrapyCommand.add_options(self, parser)
parser.add_option("--nolinks", dest="nolinks", action="store_true", help="don't show extracted links")
parser.add_option("--noitems", dest="noitems", action="store_true", help="don't show scraped items")
parser.add_option("--identify", dest="identify", action="store_true", help="try to use identify instead of parse")
parser.add_option("--nocolour", dest="nocolour", action="store_true", help="avoid using pygments to colorize the output")
def pipeline_process(self, item, opts):
return item
def run(self, args, opts):
if not args:
print "A URL is required"
return
responses = fetch([args[0]])
if responses:
response = responses[0]
spider = spiders.fromurl(response.url)
result = spider.parse(response) if not opts.identify else spider.identify(response)
items = [self.pipeline_process(i, opts) for i in result if isinstance(i, ScrapedItem)]
links = [i for i in result if isinstance(i, Request)]
display.nocolour = opts.nocolour
if not opts.noitems:
print "# Scraped Items", "-"*60
display.pprint(items)
if not opts.nolinks:
print "# Links", "-"*68
display.pprint(links)

View File

@ -0,0 +1,174 @@
import os
from scrapy.command import ScrapyCommand
from scrapy.replay import Replay
from scrapy.utils import display
from scrapy.conf import settings
class Command(ScrapyCommand):
def syntax(self):
return "[options] <replay_file> [action]"
def short_desc(self):
return "Replay a session previously recorded with crawl --record"
def help(self):
s = "Replay a session previously recorded with crawl --record\n"
s += "\n"
s += "Available actions:\n"
s += " crawl: just replay the crawl (default if action omitted)\n"
s += " diff: replay the crawl and show differences in items scraped/passed\n"
s += " update: replay the crawl and update both scraped and passed items\n"
s += " showitems: show items stored\n"
s += " showpages: show all responses downloaded (not only HTML pages)\n"
return s
def add_options(self, parser):
ScrapyCommand.add_options(self, parser)
parser.add_option("-v", "--verbose", dest="verbose", action="store_true", help="show verbose output (full items/responses)")
parser.add_option("-t", "--item-type", dest="itype", help="item type (scraped, passed). default: scraped", metavar="TYPE")
parser.add_option("--output", dest="outfile", help="write output to FILE. if omitted uses stdout", metavar="FILE")
parser.add_option("--nocolour", dest="nocolour", action="store_true", help="disable colorized output (for console only)")
parser.add_option("-i", "--ignore", dest="ignores", action="append", help="item attribute to ignore. can be passed multiple times", metavar="ATTR")
parser.add_option("--target", dest="targets", action="append", help="crawl TARGET instead of recorded urls/domains. can be passed multiple times")
def process_options(self, args, opts):
ScrapyCommand.process_options(self, args, opts)
self.opts = opts
self.action = args[1] if len(args) > 1 else 'crawl'
mode = 'update' if self.action == 'update' else 'play'
usedir = args and os.path.isdir(args[0])
self.replay = Replay(args[0], mode=mode, usedir=usedir)
if self.action not in ['crawl', 'diff', 'update']:
settings.overrides['LOG_ENABLED'] = False
def run(self, args, opts):
if not args:
print "A <replay_dir> is required"
return
display.nocolour = opts.nocolour
if opts.itype == 'passed':
self.before_db = self.replay.passed_old
self.now_db = self.replay.passed_new
else: # default is 'scraped'
opts.itype = 'scraped'
self.before_db = self.replay.scraped_old
self.now_db = self.replay.scraped_new
actionfunc = getattr(self, 'action_%s' % self.action, None)
if actionfunc:
rep = actionfunc(opts)
if rep:
if opts.outfile:
f = open(opts.outfile, "w")
f.write(rep)
f.close()
else:
print rep,
self.replay.cleanup()
else:
print "Unknown replay action: %s" % self.action
def action_crawl(self, opts):
self.replay.play(args=opts.targets)
def action_update(self, opts):
self.action_crawl(opts)
def action_showitems(self, opts):
s = ""
s += self._format_items(self.before_db.values())
s += ">>> Total: %d items %s\n" % (len(self.before_db), opts.itype)
return s
def action_showpages(self, opts):
s = ""
if self.opts.verbose:
for r in self.replay.responses_old.values():
s += ">>> %s\n" % str(r)
s += display.pformat(r)
else:
s += "\n".join([str(r) for r in self.replay.responses_old.values()]) + "\n"
s += ">>> Total: %d responses downloaded\n" % len(self.replay.responses_old)
return s
def action_diff(self, opts):
self.action_crawl(opts)
guids_before = set(self.before_db.keys())
guids_now = set(map(str, self.now_db.keys()))
guids_new = guids_now - guids_before
guids_missing = guids_before - guids_now
guids_both = guids_now & guids_before
chcount, chreport = self._report_differences(self.before_db, self.now_db, guids_both)
s = "CRAWLING DIFFERENCES REPORT\n\n"
s += "Total items : %d\n" % (len(guids_both) - chcount)
s += " Items OK : %d\n" % (len(guids_both) - chcount)
s += " New items : %d\n" % len(guids_new)
s += " Missing items : %d\n" % len(guids_missing)
s += " Changed items : %d\n" % chcount
s += "\n"
s += "- NEW ITEMS (%d) -----------------------------------------\n" % len(guids_new)
s += self._format_items([self.now_db[g] for g in guids_new])
s += "\n"
s += "- MISSING ITEMS (%d) -------------------------------------\n" % len(guids_missing)
s += self._format_items([self.before_db[g] for g in guids_missing])
s += "\n"
s += "- CHANGED ITEMS (%d) -------------------------------------\n" % chcount
s += chreport
return s
def _report_differences(self, old_items, new_items, guids):
items_old = [old_items[g] for g in guids]
items_new = [new_items[g] for g in guids]
c = 0
s = ""
for old, new in zip(items_old, items_new):
d = self._item_diff(old, new)
if d:
c += 1
s += d
return c, s
def _item_diff(self, old, new):
olddic = old.__dict__
newdic = new.__dict__
ignored_attrs = set(self.opts.ignores or [])
allattrs = [a for a in olddic.keys() + newdic.keys() if not a.startswith('_') and a not in ignored_attrs]
diff = {}
for attr in sorted(allattrs):
oldval = olddic.get(attr, None)
newval = newdic.get(attr, None)
if oldval != newval:
diff[attr] = {}
diff[attr]['old'] = oldval
diff[attr]['new'] = newval
s = ""
if diff:
s += ">>> Item guid=%s name=%s\n" % (old.guid, old.name)
s += display.pformat(diff) + "\n"
return s
def _format_items(self, items):
if self.opts.verbose:
s = display.pformat(items)
else:
s = ""
for i in items:
s += "%s\n" % str(i)
s += " <%s>\n" % i.url
return s

View File

@ -0,0 +1,81 @@
import scrapy
from scrapy.command import ScrapyCommand
from scrapy.fetcher import fetch
from scrapy.spider import spiders
from scrapy.xpath import XPathSelector
from scrapy.utils.misc import load_class
from scrapy.extension import extensions
from scrapy.conf import settings
def load_url(url, response):
vars = {}
itemcls = load_class(settings['DEFAULT_ITEM_CLASS'])
item = itemcls()
vars['item'] = item
vars['url'] = url
vars['response'] = response
vars['xs'] = XPathSelector(response)
vars['spider'] = spiders.fromurl(url)
return vars
def print_vars(vars):
print '-' * 78
print "Available local variables:"
for key, val in vars.iteritems():
if isinstance(val, basestring):
print " %s: %s" % (key, val)
else:
print " %s: %s" % (key, val.__class__)
print '-' * 78
class Command(ScrapyCommand):
def syntax(self):
return "<url>"
def short_desc(self):
return "Interactive scraping console"
def long_desc(self):
return "Interactive console for scraping the given url. For scraping local files you can use a URL like file://path/to/file.html"
def update_vars(self, vars):
""" You can use this function to update the local variables that will be available in the scrape console """
pass
def run(self, args, opts):
if not args:
return False
url = args[0]
print "Scrapy %s - Interactive scraping console\n" % scrapy.__version__
print "Enabling Scrapy extensions...",
extensions.load()
print "done"
print "Downloading URL... ",
responses = fetch([url])
print "done"
if not responses:
if not opts.loglevel:
print 'Nothing downloaded, run with -o DEBUG to see why it failed'
return
response = responses[0]
vars = load_url(url, response)
self.update_vars(vars)
print_vars(vars)
try: # use IPython if available
import IPython
shell = IPython.Shell.IPShell(argv=[], user_ns=vars)
shell.mainloop()
except ImportError:
import code
try: # readline module is only available on unix systems
import readline
except ImportError:
pass
else:
import rlcompleter
readline.parse_and_bind("tab:complete")
code.interact(local=vars)

View File

@ -0,0 +1,16 @@
from scrapy.command import ScrapyCommand
from scrapy.core.manager import scrapymanager
class Command(ScrapyCommand):
def syntax(self):
return "[options]"
def short_desc(self):
return "Start the Scrapy server"
def add_options(self, parser):
ScrapyCommand.add_options(self, parser)
def run(self, args, opts):
scrapymanager.start(*args, **opts.__dict__)

View File

@ -0,0 +1,30 @@
import pprint
from scrapy.command import ScrapyCommand
from scrapy.conf import settings
class Command(ScrapyCommand):
def syntax(self):
return "<domain> [domain ...]"
def short_desc(self):
return "Show all stats history stored for the given domain(s)"
def add_options(self, parser):
ScrapyCommand.add_options(self, parser)
parser.add_option("-p", "--path", dest="path", help="restrict stats to PATH", metavar="PATH")
def run(self, args, opts):
if not args:
print "A domain is required"
return
if not settings['SCRAPING_DB']:
print "SCRAPING_DB setting is required for this command"
return
from scrapy.store.db import DomainDataHistory
ddh = DomainDataHistory(settings['SCRAPING_DB'], 'domain_data_history')
for domain in args:
print "# %s" % domain
pprint.pprint(list(ddh.getall(domain, opts.path)))

View File

@ -0,0 +1,67 @@
"""
Base class for Scrapy commands
"""
from scrapy.conf import settings
class ScrapyCommand(object):
def syntax(self):
"""
Command syntax (preferably one-line). Do not include command name.
"""
return ""
def short_desc(self):
"""
A short description of the command
"""
return ""
def long_desc(self):
"""
A long description of the command. Return short description when not
available. It cannot contain newlines, since contents will be formatted
by optparser which removes newlines and wraps text.
"""
return self.short_desc()
def help(self):
"""
An extensive help for the command. It will be shown when using the
"help" command. It can contain newlines, since not post-formatting will
be applied to its contents.
"""
return self.long_desc()
def add_options(self, parser):
"""
Populate option parse with options available for this command
"""
parser.add_option("-f", "--logfile", dest="logfile", help="logfile to use. if omitted stderr will be used", metavar="FILE")
parser.add_option("-o", "--loglevel", dest="loglevel", default=None, help="log level")
parser.add_option("--spider", dest="spider", default=None, help="default spider (domain) to use if no spider is found")
parser.add_option("--nolog", dest="nolog", action="store_true", help="disable all log messages")
parser.add_option("--profile", dest="profile", default=None, help="write profiling stats in FILE, to analyze later with: python -m pstats FILE", metavar="FILE")
def process_options(self, args, opts):
if opts.logfile:
settings.overrides['LOG_ENABLED'] = True
settings.overrides['LOGFILE'] = opts.logfile
if opts.loglevel:
settings.overrides['LOG_ENABLED'] = True
settings.overrides['LOGLEVEL'] = opts.loglevel
if opts.nolog:
settings.overrides['LOG_ENABLED'] = False
if opts.spider:
from scrapy.spider import spiders
spiders.default_domain = opts.spider
def run(self, args, opts):
"""
Entry point for running commands
"""
raise NotImplementedError

View File

@ -0,0 +1,71 @@
import os
SETTINGS_MODULE = os.environ.get('SCRAPYSETTINGS_MODULE', 'scrapy_settings')
class Settings(object):
"""Class to obtain configuration values from settings module
which can be overriden by environment variables prepended by SCRAPY_"""
# settings in precedence order
overrides = None
settings = None
defaults = None
core = None
def __init__(self):
self.overrides = {}
self.settings = self._import(SETTINGS_MODULE)
self.defaults = {}
self.core = self._import('scrapy.conf.core_settings')
def _import(self, modulepath):
try:
return __import__(modulepath, {}, {}, [''])
except ImportError:
import sys
err = "Error: Can't find %s module in your python path\n"
sys.stderr.write(err % modulepath)
sys.exit(1)
def __getitem__(self, opt_name):
if opt_name in self.overrides:
return self.overrides[opt_name]
if 'SCRAPY_' + opt_name in os.environ:
return os.environ['SCRAPY_' + opt_name]
if hasattr(self.settings, opt_name):
return getattr(self.settings, opt_name)
if opt_name in self.defaults:
return self.defaults[opt_name]
if hasattr(self.core, opt_name):
return getattr(self.core, opt_name)
def get(self, name, default=None):
return self[name] if self[name] is not None else default
def getbool(self, name, default=False):
"""
True is: 1, '1', True
False is: 0, '0', False, None
"""
return bool(int(self.get(name, default)))
def getint(self, name, default=0):
return int(self.get(name, default))
def getfloat(self, name, default=0.0):
return float(self.get(name, default))
def getlist(self, name, default=None):
value = self.get(name)
if value is None:
return []
elif hasattr(value, '__iter__'):
return value
else:
return str(value).split(',')
settings = Settings()

View File

@ -0,0 +1 @@
LOG_STDOUT = True

View File

@ -0,0 +1 @@
LOG_ENABLED = False

View File

@ -0,0 +1 @@
LOG_ENABLED = False

View File

@ -0,0 +1 @@
LOG_ENABLED = False

View File

@ -0,0 +1,2 @@
LOG_ENABLED = False

View File

@ -0,0 +1 @@
LOG_ENABLED = False

View File

@ -0,0 +1 @@
LOG_ENABLED = False

View File

@ -0,0 +1,29 @@
import scrapy
# Scrapy core settings
BOT_NAME = 'scrapy'
BOT_VERSION = scrapy.__version__
ENGINE_DEBUG = False
# Download configuration options
USER_AGENT = '%s/%s' % (BOT_NAME, BOT_VERSION)
DOWNLOAD_TIMEOUT = 180 # 3mins
CONCURRENT_DOMAINS = 8 # number of domains to scrape in parallel
REQUESTS_PER_DOMAIN = 8 # max simultaneous requests per domain
LOG_ENABLED = True #
LOGLEVEL = 'DEBUG' # default loglevel
LOGFILE = None # None means sys.stderr by default
LOG_STDOUT = False #
DEFAULT_ITEM_CLASS = 'scrapy.item.ScrapedItem'
SCHEDULER = 'scrapy.core.scheduler.Scheduler'
MEMORYSTORE = 'scrapy.core.scheduler.MemoryStore'
PRIORITIZER = 'scrapy.core.prioritizers.RandomPrioritizer'
EXTENSIONS = []
# contrib.middleware.retry.RetryMiddleware default settings
RETRY_TIMES = 3
RETRY_HTTP_CODES = ['500', '503', '504', '400', '408', '200']

View File

View File

@ -0,0 +1,45 @@
"""
CloseDomain is an extension that forces spiders to be closed after a given
time has expired.
"""
import datetime
import pprint
from pydispatch import dispatcher
from scrapy.core import signals, log
from scrapy.core.engine import scrapyengine
from scrapy.core.mail import MailSender
from scrapy.core.exceptions import NotConfigured
from scrapy.stats import stats
from scrapy.conf import settings
class CloseDomain(object):
def __init__(self):
self.timeout = settings.getint('CLOSEDOMAIN_TIMEOUT')
if not self.timeout:
raise NotConfigured
self.tasks = {}
self.mail = MailSender()
self.notify = settings.getlist('CLOSEDOMAIN_NOTIFY')
dispatcher.connect(self.domain_opened, signal=signals.domain_opened)
dispatcher.connect(self.domain_closed, signal=signals.domain_closed)
def domain_opened(self, domain):
self.tasks[domain] = scrapyengine.addtask(self.close_domain, self.timeout, args=[domain])
def close_domain(self, domain):
log.msg("Domain was opened for more than %d seconds, closing it..." % self.timeout, domain=domain)
scrapyengine.close_domain(domain)
if self.notify:
body = "Closed domain %s because it remained opened for more than %s\n\n" % (domain, datetime.timedelta(seconds=self.timeout))
body += "DOMAIN STATS ------------------------------------------------------\n\n"
body += pprint.pformat(stats.get(domain, None))
subj = "Closed domain by timeout: %s" % domain
self.mail.send(self.notify, subj, body)
def domain_closed(self, domain):
scrapyengine.removetask(self.tasks[domain])

View File

@ -0,0 +1,2 @@
from scrapy.contrib.cluster.master import ClusterMaster, ClusterMasterWeb
from scrapy.contrib.cluster.worker import ClusterWorker, ClusterWorkerWeb

View File

@ -0,0 +1,2 @@
from scrapy.contrib.cluster.master.manager import ClusterMaster
from scrapy.contrib.cluster.master.web import ClusterMasterWeb

View File

@ -0,0 +1,189 @@
import urlparse
import urllib
import bisect
from pydispatch import dispatcher
from twisted.web.client import getPage
from scrapy.core import log, signals
from scrapy.core.engine import scrapyengine
from scrapy.core.exceptions import NotConfigured
from scrapy.utils.serialization import unserialize
from scrapy.conf import settings
class ClusterNode(object):
def __init__(self, name, url):
self.name = name
self.url = url
self.maxproc = 0
self.running = {}
self.pending = []
self.loadavg = (0.0, 0.0, 0.0)
self.status = "down" # down/crawling/idle/error
self.available = False
self.wsurl = urlparse.urljoin(self.url, "cluster_worker/ws/?format=json")
def update(self):
d = getPage(self.wsurl)
d.addCallbacks(self._cbUpdate, self._ebUpdate)
def schedule(self, domains):
args = [("schedule", domain) for domain in domains]
self._wsRequest(args)
def stop(self, domains):
args = [("stop", domain) for domain in domains]
self._wsRequest(args)
def remove(self, domains):
args = [("remove", domain) for domain in domains]
self._wsRequest(args)
def _wsRequest(self, args):
d = getPage("%s?%s" % (self.wsurl, urllib.urlencode(args)))
d.addCallbacks(self._cbUpdate, self._ebUpdate)
def _cbUpdate(self, jsonstatus):
try:
pmstatus = unserialize(jsonstatus, 'json')
self.maxproc = int(pmstatus.get('maxproc', None))
self.running = pmstatus.get('running') or {}
self.pending = pmstatus.get('pending') or []
self.loadavg = tuple(pmstatus.get('loadavg', (0.0, 0.0, 0.0)))
self.status = "crawling" if self.running else "idle"
self.available = True
except ValueError:
self.status = "error"
self.available = False
def _ebUpdate(self, err):
self.status = "down"
self.available = False
class ClusterMaster(object):
def __init__(self):
if not settings.getbool('CLUSTER_MASTER_ENABLED'):
raise NotConfigured
self.nodes = {}
dispatcher.connect(self._engine_started, signal=signals.engine_started)
def load_nodes(self):
"""Loads nodes from the CLUSTER_MASTER_NODES setting"""
for name, url in settings.get('CLUSTER_MASTER_NODES', {}).iteritems():
self.add_node(name, url)
def update_nodes(self):
for node in self.nodes.itervalues():
node.update()
def add_node(self, name, url):
"""Add node given its node"""
if name not in self.nodes:
node = ClusterNode(name, url)
self.nodes[name] = node
node.update()
def remove_node(self, nodename):
raise NotImplemented
def schedule(self, domains, nodename=None):
if nodename:
node = self.nodes[nodename]
node.schedule(domains)
else:
self._dispatch_domains(domains)
def stop(self, domains):
to_stop = {}
for domain in domains:
node = self.running.get(domain, None)
if node:
if node.name not in to_stop:
to_stop[node.name] = []
to_stop[node.name].append(domain)
for nodename, domains in to_stop.iteritems():
self.nodes[nodename].stop(domains)
def remove(self, domains):
to_remove = {}
for domain in domains:
node = self.pending.get(domain, None)
if node:
if node.name not in to_remove:
to_remove[node.name] = []
to_remove[node.name].append(domain)
for nodename, domains in to_remove.iteritems():
self.nodes[nodename].remove(domains)
def discard(self, domains):
"""Stop and remove all running and pending instances of the given
domains"""
self.remove(domains)
self.stop(domains)
@property
def running(self):
"""Return dict of running domains as domain -> node"""
d = {}
for node in self.nodes.itervalues():
for domain in node.running.iterkeys():
d[domain] = node
return d
@property
def pending(self):
"""Return dict of pending domains as domain -> node"""
d = {}
for node in self.nodes.itervalues():
for domain in node.pending:
d[domain] = node
return d
@property
def available_nodes(self):
return (node for node in self.nodes.itervalues() if node.available)
def _dispatch_domains(self, domains):
"""Schedule the given domains in the availables nodes as good as
possible. The algorithm follows the next rules (in order):
1. search for nodes with available capacity(running < maxproc) and (if
any) schedules the domains there
2. if there isn't any node with available capacity it schedules the
domain in the node with the smallest number of pending spiders
"""
to_schedule = {} # domains to schedule per node
pending_node = [] # list of #pending, node
for node in self.available_nodes:
capacity = node.maxproc - len(node.running)
bisect.insort(pending_node, (len(node.pending), node))
to_schedule[node.name] = []
while domains and capacity > 0:
to_schedule[node.name].append(domains.pop(0))
capacity -= 1
if not domains:
break
for domain in domains:
pending, node = pending_node.pop(0)
to_schedule[node.name].append(domain)
bisect.insort(pending_node, (pending+1, node))
for nodename, domains in to_schedule.iteritems():
if domains:
self.nodes[nodename].schedule(domains)
def _engine_started(self):
self.load_nodes()
scrapyengine.addtask(self.update_nodes, settings.getint('CLUSTER_MASTER_POLL_INTERVAL'))

View File

@ -0,0 +1,165 @@
import datetime
from pydispatch import dispatcher
from scrapy.spider import spiders
from scrapy.management.web import banner, webconsole_discover_module
from scrapy.utils.serialization import parse_jsondatetime
from scrapy.contrib.cluster.master import ClusterMaster
class ClusterMasterWeb(ClusterMaster):
webconsole_id = 'cluster_master'
webconsole_name = 'Cluster master'
def __init__(self):
ClusterMaster.__init__(self)
dispatcher.connect(self.webconsole_discover_module, signal=webconsole_discover_module)
def webconsole_render(self, wc_request):
changes = ""
if wc_request.path == '/cluster/nodes/':
return self.render_nodes(wc_request)
elif wc_request.path == '/cluster/domains/':
return self.render_domains(wc_request)
elif wc_request.args:
changes = self.webconsole_control(wc_request)
s = self.render_header()
s += "<h2>Home</h2>\n"
s += "<table border='1'>\n"
s += "<tr><th>&nbsp;</th><th>Name</th><th>Status</th><th>Running</th><th>Pending</th><th>Load.avg</th></tr>\n"
for node in self.nodes.itervalues():
#chkbox = "<input type='checkbox' name='shutdown' value='%s' />" % domain if node.status in ["up", "idle"] else "&nbsp;"
nodelink = "<a href='nodes/#%s'>%s</a>" % (node.name, node.name)
chkbox = "&nbsp;"
loadavg = "%.2f %.2f %.2f" % node.loadavg
s += "<tr><td>%s</td><td>%s</td><td>%s</td><td>%d/%d</td><td>%d</td><td>%s</td></tr>\n" % \
(chkbox, nodelink, node.status, len(node.running), node.maxproc, len(node.pending), loadavg)
s += "</table>\n"
s += "</body>\n"
s += "</html>\n"
return str(s)
def webconsole_control(self, wc_request):
args = wc_request.args
if "updatenodes" in args:
self.update_nodes()
if "schedule" in args:
node = args["node"][0] if "node" in args else None
self.schedule(args["schedule"], nodename=node)
if "stop" in args:
self.stop(args["stop"])
if "remove" in args:
self.remove(args["remove"])
return ""
def render_nodes(self, wc_request):
if wc_request.args:
self.webconsole_control(wc_request)
now = datetime.datetime.utcnow()
s = self.render_header()
for node in self.nodes.itervalues():
s += "<h2><a name='%s'>%s</h2>\n" % (node.name, node.name)
s += "<h3>Running domains</h3>\n"
if node.running:
s += "<form method='post' action='.'>\n"
s += "<table border='1'>\n"
s += "<tr><th>&nbsp;</th><th>PID</th><th>Domain</th><th>Status</th><th>Running time</th><th>Log file</th></tr>\n"
for domain, proc in node.running.iteritems():
chkbox = "<input type='checkbox' name='stop' value='%s' />" % domain if proc['status'] == "running" else "&nbsp;"
start_time = parse_jsondatetime(proc.get('start_time', None))
elapsed = now - start_time if start_time else None
s += "<tr><td>%s</td><td>%s</td><td>%s</td><td>%s</td><td>%s</td><td>%s</td></tr>\n" % \
(chkbox, proc['pid'], domain, proc['status'], elapsed, proc['logfile'])
s += "</table>\n"
s += "<input type='hidden' name='node' value='%s'>\n" % node.name
s += "<p><input type='submit' value='Stop selected domains on %s'></p>\n" % node.name
s += "</form>\n"
else:
s += "<p>No running domains on %s</p>\n" % node.name
# pending domains
s += "<h3>Pending domains</h3>\n"
if node.pending:
s += "<form method='post' action='.'>\n"
s += "<select name='remove' multiple='multiple' size='10'>\n"
for domain in node.pending:
s += "<option>%s</option>\n" % domain
s += "</select>\n"
s += "<input type='hidden' name='node' value='%s'>\n" % node.name
s += "<p><input type='submit' value='Remove selected pending domains on %s'></p>\n" % node.name
s += "</form>\n"
else:
s += "<p>No pending domains on %s</p>\n" % node.name
return str(s)
def render_domains(self, wc_request):
if wc_request.args:
self.webconsole_control(wc_request)
enabled_domains = set(spiders.asdict(include_disabled=False).keys())
inactive_domains = enabled_domains - set(self.running.keys() + self.pending.keys())
s = self.render_header()
s += "<h2>Schedule domains</h2>\n"
s += "Inactive domains (not running or pending)<br />"
s += "<form method='post' action='.'>\n"
s += "<select name='schedule' multiple='multiple' size='10'>\n"
for domain in sorted(inactive_domains):
s += "<option>%s</option>\n" % domain
s += "</select>\n"
s += "</br>\n"
s += "Node (only available nodes shown):<br />\n"
s += "<select name='node'>\n"
s += "<option value='' selected='selected'>any</option>"
for node in self.available_nodes:
domcount = "%d/%d/%d" % (len(node.running), node.maxproc, len(node.pending))
loadavg = "%.2f %.2f %.2f" % node.loadavg
s += "<option value='%s'>%s [D: %s | LA: %s]</option>" % (node.name, node.name, domcount, loadavg)
s += "</select>\n"
s += "<p><input type='submit' value='Schedule selected domains'></p>\n"
s += "</form>\n"
s += "<h2>Domains</h2>\n"
s += "<table border='1'>\n"
s += "<tr><th>Domain</th><th>Status</th><th>Node</th></tr>\n"
s += self._domains_table(self.running, '<b>running</b>')
s += self._domains_table(self.pending, 'pending')
s += "</table>\n"
return str(s)
def render_header(self):
s = banner(self)
s += "<p>Nav: "
s += "<a href='/cluster/'>Home</a> | "
s += "<a href='/cluster/domains/'>Domains</a> | "
s += "<a href='/cluster/nodes/'>Nodes</a> (<a href='/cluster/nodes/?updatenodes=1'>update</a>)"
s += "</p>"
return s
def _domains_table(self, dict_, status):
s = ""
for domain, node in dict_.iteritems():
s += "<tr><td>%s</td><td>%s</td><td>%s</td></tr>\n" % (domain, status, node.name)
return s
def webconsole_discover_module(self):
return self

View File

@ -0,0 +1,2 @@
from scrapy.contrib.cluster.worker.manager import ClusterWorker
from scrapy.contrib.cluster.worker.web import ClusterWorkerWeb

View File

@ -0,0 +1,96 @@
import sys
import os
import time
import datetime
from twisted.internet import protocol, reactor
from scrapy.core import log
from scrapy.core.exceptions import NotConfigured
from scrapy.conf import settings
class ScrapyProcessProtocol(protocol.ProcessProtocol):
def __init__(self, procman, domain, logfile=None):
self.procman = procman
self.domain = domain
self.logfile = logfile
self.start_time = datetime.datetime.utcnow()
self.status = "starting"
self.pid = -1
def __str__(self):
return "<ScrapyProcess domain=%s, pid=%s, status=%s>" % (self.domain, self.pid, self.status)
def connectionMade(self):
self.pid = self.transport.pid
log.msg("ClusterWorker: started domain=%s, pid=%d, log=%s" % (self.domain, self.pid, self.logfile))
self.transport.closeStdin()
self.status = "running"
def processEnded(self, status_object):
log.msg("ClusterWorker: finished domain=%s, pid=%d, log=%s" % (self.domain, self.pid, self.logfile))
del self.procman.running[self.domain]
self.procman.next_pending()
class ClusterWorker(object):
def __init__(self):
if not settings.getbool('CLUSTER_WORKER_ENABLED'):
raise NotConfigured
self.maxproc = settings.getint('CLUSTER_WORKER_MAXPROC')
self.logdir = settings['CLUSTER_WORKER_LOGDIR']
self.running = {}
self.pending = []
def schedule(self, domain):
"""Schedule new domain to be crawled in a separate processes"""
if len(self.running) < self.maxproc and domain not in self.running:
self._run(domain)
else:
self.pending.append(domain)
def stop(self, domain):
"""Stop running domain. For removing pending (not yet started) domains
use remove() instead"""
if domain in self.running:
proc = self.running[domain]
log.msg("ClusterWorker: Sending shutdown signal to domain=%s, pid=%d" % (domain, proc.pid))
proc.transport.signalProcess('INT')
proc.status = "closing"
def remove(self, domain):
"""Remove all scheduled instances of the given domain (if it hasn't
started yet). Otherwise use stop()"""
while domain in self.pending:
self.pending.remove(domain)
def next_pending(self):
"""Run the next domain in the pending list, which is not already running"""
if len(self.running) >= self.maxproc:
return
for domain in self.pending:
if domain not in self.running:
self._run(domain)
self.pending.remove(domain)
return
def _run(self, domain):
"""Spawn process to run the given domain. Don't call this method
directly. Instead use schedule()."""
logfile = os.path.join(self.logdir, domain, time.strftime("%FT%T.log"))
if not os.path.exists(os.path.dirname(logfile)):
os.makedirs(os.path.dirname(logfile))
scrapy_proc = ScrapyProcessProtocol(self, domain, logfile)
env = {'SCRAPY_LOGFILE': logfile}
args = [sys.executable, sys.argv[0], 'crawl', domain]
proc = reactor.spawnProcess(scrapy_proc, sys.executable, args=args, env=env)
self.running[domain] = scrapy_proc

View File

@ -0,0 +1,127 @@
import os
import datetime
from pydispatch import dispatcher
from scrapy.spider import spiders
from scrapy.management.web import banner, webconsole_discover_module
from scrapy.utils.serialization import serialize
from scrapy.contrib.cluster.worker import ClusterWorker
class ClusterWorkerWeb(ClusterWorker):
webconsole_id = 'cluster_worker'
webconsole_name = 'Cluster worker'
def __init__(self):
ClusterWorker.__init__(self)
dispatcher.connect(self.webconsole_discover_module, signal=webconsole_discover_module)
def webconsole_render(self, wc_request):
changes = ""
if wc_request.path == '/cluster_worker/ws/':
return self.webconsole_control(wc_request, ws=True)
elif wc_request.args:
changes = self.webconsole_control(wc_request)
now = datetime.datetime.utcnow()
s = banner(self)
# running processes
s += "<h2>Running processes</h2>\n"
if self.running:
s += "<form method='post' action='.'>\n"
s += "<table border='1'>\n"
s += "<tr><th>&nbsp;</th><th>PID</th><th>Domain</th><th>Status</th><th>Log file</th><th>Running time</th></tr>\n"
for domain, proc in self.running.iteritems():
chkbox = "<input type='checkbox' name='stop' value='%s' />" % domain if proc.status == "running" else "&nbsp;"
elapsed = now - proc.start_time
s += "<tr><td>%s</td><td>%s</td><td>%s</td><td>%s</td><td>%s</td><td>%s</td></tr>\n" % \
(chkbox, proc.pid, domain, proc.status, proc.logfile, elapsed)
s += "</table>\n"
s += "<p><input type='submit' value='Stop selected domains'></p>\n"
s += "</form>\n"
else:
s += "<p>No running processes</p>\n"
# pending domains
s += "<h2>Pending domains</h2>\n"
if self.pending:
s += "<form method='post' action='.'>\n"
s += "<select name='remove' multiple='multiple' size='10'>\n"
for domain in self.pending:
s += "<option>%s</option>\n" % domain
s += "</select>\n"
s += "<p><input type='submit' value='Remove selected pending domains'></p>\n"
s += "</form>\n"
else:
s += "<p>No pending domains</p>\n"
# schedule domains
enabled_domains = spiders.asdict(include_disabled=False).keys()
s += "<h2>Schedule domains</h2>\n"
s += "<form method='post' action='.'>\n"
s += "<select name='schedule' multiple='multiple' size='10'>\n"
for domain in sorted(enabled_domains):
s += "<option>%s</option>\n" % domain
s += "</select>\n"
s += "<p><input type='submit' value='Schedule selected domains'></p>\n"
s += "</form>\n"
s += changes
s += "</body>\n"
s += "</html>\n"
return s
def webconsole_control(self, wc_request, ws=False):
args = wc_request.args
if "schedule" in args:
for domain in args["schedule"]:
self.schedule(domain)
if ws:
return self.ws_status(wc_request)
else:
return "<p>Scheduled domains: <ul><li>%s</li></ul>" % "</li><li>".join(args["schedule"]) + "</p>\n"
if "stop" in args:
for domain in args["stop"]:
self.stop(domain)
if ws:
return self.ws_status(wc_request)
else:
return "<p>Stopped running processes: <ul><li>%s</li></ul>" % "</li><li>".join(args["stop"]) + "</p>\n"
if "remove" in args:
for domain in args["remove"]:
self.remove(domain)
if ws:
return self.ws_status(wc_request)
else:
return "<p>Removed pending domains: <ul><li>%s</li></ul>" % "</li><li>".join(args["remove"]) + "</p>\n"
if ws:
return self.ws_status(wc_request)
else:
return ""
def ws_status(self, wc_request):
format = wc_request.args['format'][0] if 'format' in wc_request.args else 'json'
wc_request.setHeader('content-type', 'text/plain')
exported_proc_attrs = ['pid', 'status', 'start_time', 'logfile']
d = {'maxproc': self.maxproc, 'running': {}, 'pending': []}
for domain, proc in self.running.iteritems():
d2 = {}
for a in exported_proc_attrs:
d2[a] = getattr(proc, a)
d['running'][domain] = d2
d['pending'] = self.pending
d['loadavg'] = os.getloadavg()
content = serialize(d, format)
return content
def webconsole_discover_module(self):
return self

View File

@ -0,0 +1,17 @@
"""
Extensions for debugging Scrapy
"""
import signal
import traceback
class StackTraceDebug(object):
def __init__(self):
try:
signal.signal(signal.SIGUSR1, self.dump_stacktrace)
except AttributeError:
# win32 platforms don't support SIGUSR signals
pass
def dump_stacktrace(self, signum, frame):
print "Got signal. Dumping stack trace..."
traceback.print_stack(frame)

View File

@ -0,0 +1,173 @@
from __future__ import with_statement
import os
import sha
import datetime
import urlparse
from pydispatch import dispatcher
from twisted.internet import defer
from scrapy.core import signals, log
from scrapy.core.engine import scrapyengine
from scrapy.http import Response, Headers
from scrapy.http.headers import headers_dict_to_raw
from scrapy.core.exceptions import UsageError, NotConfigured, HttpException, IgnoreRequest
from scrapy.conf import settings
class CacheMiddleware(object):
def __init__(self):
if not settings['CACHE2_DIR']:
raise NotConfigured
self.cache = Cache(settings['CACHE2_DIR'], sectorize=settings.getbool('CACHE2_SECTORIZE'))
self.ignore_missing = settings.getbool('CACHE2_IGNORE_MISSING')
dispatcher.connect(self.open_domain, signal=signals.domain_open)
def open_domain(self, domain):
self.cache.open_domain(domain)
def process_request(self, request, spider):
if not is_cacheable(request):
return
key = request.fingerprint()
domain = spider.domain_name
response = self.cache.retrieve_response(domain, key)
if response:
response.cached = True
if not 200 <= int(response.status) < 300:
raise HttpException(response.status, None, response)
return response
elif self.ignore_missing:
raise IgnoreRequest("Ignored request not in cache: %s" % request)
def process_response(self, request, response, spider):
if not is_cacheable(request):
return response
if isinstance(response, Response) and not response.cached:
key = request.fingerprint()
domain = spider.domain_name
self.cache.store(domain, key, request, response)
return response
def process_exception(self, request, exception, spider):
if not is_cacheable(request):
return
if isinstance(exception, HttpException) and isinstance(exception.response, Response):
key = request.fingerprint()
domain = spider.domain_name
self.cache.store(domain,key, request, exception.response)
def is_cacheable(request):
scheme, _, _, _, _ = urlparse.urlsplit(request.url)
return scheme in ['http', 'https']
class Cache(object):
DOMAIN_SECTORDIR = 'data'
DOMAIN_LINKDIR = 'domains'
def __init__(self, cachedir, sectorize=False):
self.cachedir = cachedir
self.sectorize = sectorize
self.baselinkpath = os.path.join(self.cachedir, self.DOMAIN_LINKDIR)
if not os.path.exists(self.baselinkpath):
os.makedirs(self.baselinkpath)
self.basesectorpath = os.path.join(self.cachedir, self.DOMAIN_SECTORDIR)
if not os.path.exists(self.basesectorpath):
os.makedirs(self.basesectorpath)
def domainsectorpath(self, domain):
sector = sha.sha(domain).hexdigest()[0]
return os.path.join(self.basesectorpath, sector, domain)
def domainlinkpath(self, domain):
return os.path.join(self.baselinkpath, domain)
def requestpath(self, domain, key):
linkpath = self.domainlinkpath(domain)
return os.path.join(linkpath, key[0:2], key)
def open_domain(self, domain):
if domain:
linkpath = self.domainlinkpath(domain)
if self.sectorize:
sectorpath = self.domainsectorpath(domain)
if not os.path.exists(sectorpath):
os.makedirs(sectorpath)
if not os.path.exists(linkpath):
try:
os.symlink(sectorpath, linkpath)
except:
os.makedirs(linkpath) # windows filesystem
else:
if not os.path.exists(linkpath):
os.makedirs(linkpath)
def is_cached(self, domain, key):
requestpath = self.requestpath(domain, key)
return os.path.exists(requestpath)
def retrieve_response(self, domain, key):
"""
Return response dictionary if request has correspondent cache record;
return None if not.
"""
requestpath = self.requestpath(domain, key)
if not os.path.exists(requestpath):
return None # not cached
metadata = responsebody = responseheaders = None
with open(os.path.join(requestpath, 'meta_data')) as f:
metadata = f.read()
with open(os.path.join(requestpath, 'response_body')) as f:
responsebody = f.read()
with open(os.path.join(requestpath, 'response_headers')) as f:
responseheaders = f.read()
metadata = eval(metadata)
url = metadata['url']
original_url = metadata.get('original_url', url)
headers = Headers(responseheaders)
parent = metadata.get('parent')
status = metadata['status']
response = Response(domain=domain, url=url, original_url=original_url, headers=headers, status=status, body=responsebody, parent=parent)
response.cached = True
return response
def store(self, domain, key, request, response):
requestpath = self.requestpath(domain, key)
if not os.path.exists(requestpath):
os.makedirs(requestpath)
metadata = {
'url':request.url,
'method': request.method,
'parent': response.parent,
'status': response.status,
'domain': response.domain,
'original_url': response.original_url,
'timestamp': datetime.datetime.now(),
}
# metadata
with open(os.path.join(requestpath, 'meta_data'), 'w') as f:
f.write(repr(metadata))
# response
with open(os.path.join(requestpath, 'response_headers'), 'w') as f:
f.write(headers_dict_to_raw(response.headers))
with open(os.path.join(requestpath, 'response_body'), 'w') as f:
f.write(response.body.get_content())
# request
with open(os.path.join(requestpath, 'request_headers'), 'w') as f:
f.write(headers_dict_to_raw(request.headers))
if request.body:
with open(os.path.join(requestpath, 'request_body'), 'w') as f:
f.write(request.body)

View File

@ -0,0 +1,10 @@
class CommonMiddleware(object):
"""This middleware provides common/basic functionality, and should always
be enabled"""
def process_request(self, request, spider):
request.headers.setdefault('Accept-Language', 'en')
request.headers.setdefault('Accept', 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8')
if request.method == 'POST':
request.headers.setdefault('Content-Type', 'application/x-www-form-urlencoded')

View File

@ -0,0 +1,42 @@
import zlib
from gzip import GzipFile
from cStringIO import StringIO
from scrapy.http import Response, ResponseBody
class CompressionMiddleware(object):
"""This middleware allows compressed (gzip, deflate) traffic to be
sent/received from web sites"""
def process_request(self, request, spider):
request.headers.setdefault('Accept-Encoding', 'gzip,deflate')
def process_response(self, request, response, spider):
if isinstance(response, Response):
content_encoding = response.headers.get('Content-Encoding')
if content_encoding:
encoding = content_encoding[0].lower()
raw_body = response.body.get_content()
declared_encoding = response.body.declared_encoding
decoded_body = self._decode(raw_body, encoding)
response.body = ResponseBody(decoded_body, declared_encoding)
response.headers['Content-Encoding'] = content_encoding[1:]
return response
def _decode(self, body, encoding):
if encoding == 'gzip':
body = GzipFile(fileobj=StringIO(body)).read()
if encoding == 'deflate':
try:
body = zlib.decompress(body)
except zlib.error:
# ugly hack to work with raw deflate content that may
# be sent by microsof servers. For more information, see:
# http://carsten.codimi.de/gzip.yaws/
# http://www.port80software.com/200ok/archive/2005/10/31/868.aspx
# http://www.gzip.org/zlib/zlib_faq.html#faq38
body = zlib.decompress(body, -15)
return body

View File

@ -0,0 +1,27 @@
from pydispatch import dispatcher
from scrapy.core import signals
from scrapy.core.engine import scrapyengine
from scrapy.utils.misc import dict_updatedefault
class CookiesMiddleware(object):
"""This middleware enables working with sites that need cookies"""
def __init__(self):
self.cookies = {}
dispatcher.connect(self.domain_open, signals.domain_open)
dispatcher.connect(self.domain_closed, signals.domain_closed)
def process_request(self, request, spider):
dict_updatedefault(request.cookies, self.cookies[spider.domain_name])
def process_response(self, request, response, spider):
cookies = self.cookies[spider.domain_name]
cookies.update(request.cookies)
return response
def domain_open(self, domain):
self.cookies[domain] = {}
def domain_closed(self, domain):
del self.cookies[domain]

View File

@ -0,0 +1,20 @@
from scrapy.core import log
from scrapy.conf import settings
class CrawlDebug(object):
def __init__(self):
self.enabled = settings.getbool('CRAWL_DEBUG')
def process_request(self, request, spider):
if self.enabled:
log.msg("Crawling %s" % repr(request), domain=spider.domain_name, level=log.DEBUG)
def process_exception(self, request, exception, spider):
if self.enabled:
log.msg("Crawl exception %s in %s" % (exception, repr(request)), domain=spider.domain_name, level=log.DEBUG)
def process_response(self, request, response, spider):
if self.enabled:
log.msg("Fetched %s from %s" % (response.info(), repr(request)), domain=spider.domain_name, level=log.DEBUG)
return response

View File

@ -0,0 +1,13 @@
from scrapy.core.exceptions import HttpException
class ErrorPagesMiddleware(object):
"""This middleware allows the spiders to receive error (non 200) responses,
the same way the receive normal responses"""
def process_exception(self, request, exception, spider):
if isinstance(exception, HttpException):
statuses = getattr(spider, 'handle_httpstatus_list', None)
httpstatus = exception.response.status
if statuses and httpstatus in statuses:
return exception.response

View File

@ -0,0 +1,8 @@
class HttpAuthMiddleware(object):
"""This middleware allows spiders to use HTTP auth in a cleaner way
(http_user and http_pass spider class attributes)"""
def process_request(self, request, spider):
if getattr(spider, 'http_user', None) or getattr(spider, 'http_pass', None):
request.httpauth(spider.http_user, spider.http_pass)

View File

@ -0,0 +1,55 @@
import re
from scrapy.core import log
from scrapy.http import Request, Response
from scrapy.core.exceptions import HttpException
from scrapy.utils.url import urljoin_rfc as urljoin
META_REFRESH_RE = re.compile(r'<meta[^>]*http-equiv[^>]*refresh[^>].*?(\d+);url=([^"\']+)', re.IGNORECASE)
# some sites use meta-refresh for redirecting to a session expired page, so we
# restrict automatic redirection to a maximum delay (in number of seconds)
META_REFRESH_MAXSEC = 100
class RedirectMiddleware(object):
def process_exception(self, request, exception, spider):
if isinstance(exception, HttpException):
status = exception.status
response = exception.response
if status in ['302', '303']:
redirected_url = urljoin(request.url, response.headers['location'][0])
if not getattr(spider, "no_redirect", False):
redirected = request.copy()
redirected.url = redirected_url
redirected.method = 'GET'
redirected.body = None
# This is needed to avoid redirection loops with requests that contain dont_filter = True
# Example (9 May 2008): http://www.55max.com/product/001_photography.asp?3233,0,0,0,Michael+Banks
redirected.dont_filter = False
log.msg("Redirecting (%s) to %s from %s" % (status, redirected, request), level=log.DEBUG, domain=spider.domain_name)
return redirected
log.msg("Ignored redirecting (%s) to %s from %s (disabled by spider)" % (status, redirected_url, request), level=log.DEBUG, domain=spider.domain_name)
return response
if status in ['301', '307']:
redirected_url = urljoin(request.url, response.headers['location'][0])
if not getattr(spider, "no_redirect", False):
redirected = request.copy()
redirected.url = redirected_url
# This is needed to avoid redirection loops with requests that contain dont_filter = True
# Example (9 May 2008): http://www.55max.com/product/001_photography.asp?3233,0,0,0,Michael+Banks
redirected.dont_filter = False
log.msg("Redirecting (%s) to %s from %s" % (status, redirected, request), level=log.DEBUG, domain=spider.domain_name)
return redirected
log.msg("Ignored redirecting (%s) to %s from %s (disabled by spider)" % (status, redirected_url, request), level=log.DEBUG, domain=spider.domain_name)
return response
def process_response(self, request, response, spider):
if isinstance(response, Response):
m = META_REFRESH_RE.search(response.body.to_string()[0:4096])
if m and int(m.group(1)) < META_REFRESH_MAXSEC:
redirected = request.copy()
redirected.url = urljoin(request.url, m.group(2))
log.msg("Redirecting (meta refresh) to %s from %s" % (redirected, request), level=log.DEBUG, domain=spider.domain_name)
return redirected
return response

View File

@ -0,0 +1,63 @@
"""
An extension to retry failed requests that are potentially caused by temporary
problems such as a connection timeout or HTTP 500 error.
You can change the behaviour of this moddileware by modifing the scraping settings:
RETRY_TIMES - how many times to retry a failed page
RETRY_HTTP_CODES - which HTTP response codes to retry
Failed pages are collected on the scraping process and rescheduled at the end,
once the spider has finished crawling all regular (non failed) pages. Once
there is no more failed pages to retry this middleware sends a signal
(retry_complete), so other extensions could connect to that signal.
Default values are located in scrapy.conf.core_settings
About HTTP errors to consider:
- You may want to remove 400 from RETRY_HTTP_CODES, if you stick to the HTTP
protocol. It's included by default because it's a common code used to
indicate server overload, which would be something we want to retry
- 200 is included by default (and shoudln't be removed) to check for partial
downloads errors, which means the TCP connection has broken in the middle of
a HTTP download
"""
from twisted.internet.error import TimeoutError as ServerTimeoutError, DNSLookupError, \
ConnectionRefusedError, ConnectionDone, ConnectError
from twisted.internet.defer import TimeoutError as UserTimeoutError
from scrapy.core import log
from scrapy.core.exceptions import HttpException
from scrapy.conf import settings
class RetryMiddleware(object):
EXCEPTIONS_TO_RETRY = (ServerTimeoutError, UserTimeoutError, DNSLookupError,
ConnectionRefusedError, ConnectionDone, ConnectError)
def __init__(self):
self.failed_count = {}
self.max_retries = settings.getint('RETRY_TIMES')
def process_exception(self, request, exception, spider):
retry = False
if isinstance(exception, self.EXCEPTIONS_TO_RETRY):
retry = True
elif isinstance(exception, HttpException):
if exception.status in settings.getlist('RETRY_HTTP_CODES'):
retry = True
if retry:
fp = request.fingerprint()
count = self.failed_count[fp] = self.failed_count.get(fp, 0) + 1
if self.failed_count[fp] < self.max_retries:
log.msg("Retrying %s (failed %d times): %s" % (request, count, exception), level=log.DEBUG, domain=spider.domain_name)
retryreq = request.copy()
retryreq.dont_filter = True
return retryreq
else:
log.msg("Discarding %s (failed %d times): %s" % (request, count, exception), domain=spider.domain_name, level=log.DEBUG)

View File

@ -0,0 +1,59 @@
"""
jhis is an expertimental middleware to respect robots.txt policies. The biggest
problem it has is that it uses urllib directly (in RobotFileParser.read()
method) and that conflicts with twisted networking, so it should be ported to
use twisted networking API, but that is not as trivial as it may seem.
This code is left here for future reference, when we resume the work on this
subject.
"""
import re
import urlparse
import robotparser
from pydispatch import dispatcher
from scrapy.core import log, signals
from scrapy.core.exceptions import IgnoreRequest
from scrapy.conf import settings
BASEURL_RE = re.compile("http://.*?/")
class RobotsMiddleware(object):
def __init__(self):
self._parsers = {}
self._spiderdomains = {}
self._pending = {}
dispatcher.connect(self.domain_open, signals.domain_open)
dispatcher.connect(self.domain_closed, signals.domain_closed)
def process_request(self, request, spider):
agent = getattr(spider, 'user_agent', None) or settings['USER_AGENT']
rp = self.robot_parser(request.url, spider.domain_name)
if rp and not rp.can_fetch(agent, request.url):
raise IgnoreRequest("URL forbidden by robots.txt: %s" % request.url)
def robot_parser(self, url, spiderdomain):
urldomain = urlparse.urlparse(url).hostname
if urldomain in self._parsers:
rp = self._parsers[urldomain]
else:
rp = robotparser.RobotFileParser()
m = BASEURL_RE.search(url)
if m:
rp.set_url("%srobots.txt" % m.group())
rp.read()
self._parsers[urldomain] = rp
self._spiderdomains[spiderdomain].add(urldomain)
return rp
def domain_open(self, domain):
self._spiderdomains[domain] = set()
def domain_closed(self, domain):
for urldomain in self._spiderdomains[domain]:
del self._parsers[urldomain]
del self._spiderdomains[domain]

View File

@ -0,0 +1,7 @@
class UserAgentMiddleware(object):
"""This middleware allows spiders to override the user_agent"""
def process_request(self, request, spider):
if getattr(spider, 'user_agent', None):
request.headers.setdefault('User-Agent', spider.user_agent)

View File

@ -0,0 +1,26 @@
"""
Extensions to override scrapy settings with per-group settings according to the
group the spider belongs to. It only overrides the settings when running the
crawl command with *only one domain as argument*.
"""
from scrapy.conf import settings
from scrapy.core.exceptions import NotConfigured
from scrapy.command.cmdline import command_executed
class GroupSettings(object):
def __init__(self):
if not settings.getbool("GROUPSETTINGS_ENABLED"):
raise NotConfigured
if command_executed and command_executed['name'] == 'crawl':
mod = __import__(settings['GROUPSETTINGS_MODULE'], {}, {}, [''])
args = command_executed['args']
if len(args) == 1 and not args[0].startswith('http://'):
domain = args[0]
settings.overrides.update(mod.default_settings)
for group, domains in mod.group_spiders.iteritems():
if domain in domains:
settings.overrides.update(mod.group_settings.get(group, {}))

View File

@ -0,0 +1,3 @@
from scrapy.contrib.history.history import ItemHistory
from scrapy.contrib.history.scheduler import RulesScheduler
from scrapy.contrib.history.store import SQLHistoryStore

View File

@ -0,0 +1,162 @@
"""
History management
"""
import re
from datetime import datetime
import MySQLdb
class History(object):
"""Base class for tracking different kinds of histories"""
def __init__(self, db_uri):
self.db_uri = db_uri
self._mysql_conn = None
def connect(self):
"""
Connect to PDB and open mysql connect to PRODUCT_DB
"""
m = re.search(r"mysql:\/\/(?P<user>[^:]+)(:(?P<passwd>[^@]+))?@(?P<host>[^/]+)/(?P<db>.*)$", self.db_uri)
if m:
d = m.groupdict()
if d['passwd'] is None:
del(d['passwd'])
d['charset'] = "utf8"
self._mysql_conn = MySQLdb.connect(**d)
def get_mysql_conn(self):
if self._mysql_conn is None:
self.connect()
return self._mysql_conn
mysql_conn = property(get_mysql_conn)
class ItemHistory(History):
"""
Class instance registers item guids, and versions with check_item method.
It also gives access to stored ItemTicket and ItemVersion objects.
"""
NEW = 0
UPDATE = 1
DUPLICATE = 2
def check_item(self, domain, item):
"""
Store item's guid and version to the database along
with date and time of last occurence.
Return:
* ItemHistory.NEW - if guid hasn't been met
* ItemHistory.UPDATE - if history already contains guid, but versions doesn't match
* ItemHistory.DUPLICATE - both guid and version are not new
"""
version = item.version
c = self.mysql_conn.cursor(MySQLdb.cursors.DictCursor)
def add_version(version):
insert = "INSERT INTO version (guid, version, seen) VALUES (%s,%s,%s)"
c.execute(insert, (item.guid, version, datetime.now()))
select = "SELECT * FROM ticket WHERE guid=%s"
c.execute(select, item.guid)
r = c.fetchone()
if r:
select = "SELECT * FROM version WHERE version=%s"
if c.execute(select, version):
update = "UPDATE version SET seen=%s WHERE version=%s"
c.execute(update, (datetime.now(), version))
self.mysql_conn.commit()
return ItemHistory.DUPLICATE
else:
add_version(version)
self.mysql_conn.commit()
return ItemHistory.UPDATE
else:
insert = "INSERT INTO ticket (guid, domain, url, url_hash) VALUES (%s,%s,%s,%s)"
c.execute(insert, (item.guid, domain, item.url, hash(item.url)))
add_version(version)
self.mysql_conn.commit()
return ItemHistory.NEW
def get_ticket(self, guid):
"""
Return ItemTicket object for guid.
ItemVersion objects can be accessed via 'versions' list.
"""
c = self.mysql_conn.cursor(MySQLdb.cursors.DictCursor)
select = "SELECT * FROM ticket WHERE guid=%s"
c.execute(select, guid)
ticket = c.fetchone()
if not ticket:
raise Exception("Item ticket with guid = '%s' not found" % guid)
ticket['versions'] = []
select = "SELECT * FROM version WHERE guid=%s"
c.execute(select, guid)
for version in c.fetchall():
ticket['versions'].append(version)
return ticket
def delete_ticket(self, guid):
"""Delete item ticket and associated versions from DB"""
c = self.mysql_conn.cursor()
delete = "DELETE FROM ticket WHERE guid=%s"
c.execute(delete, guid)
self.mysql_conn.commit()
class URLHistory(History):
"""
Access URL status and history for the scraping engine
This is degsigned to have an instance per domain where typically
a call will be made to get_url_status, followed by either
update_checked or record_version.
"""
def get_url_status(self, urlkey):
"""
Get the url status (url, last_version, last_checked),
or None if the url data has not been seen before
"""
c = self.mysql_conn.cursor(MySQLdb.cursors.DictCursor)
select = "SELECT * FROM url_status WHERE url_hash=%s"
c.execute(select, urlkey)
r = c.fetchone()
return (r['url'], r['last_version'], r['last_checked']) if r else None
def record_version(self, urlkey, url, parent_key, version, postdata_hash=None):
"""
Record a version of a page and update the last checked time.
If the same version (or None) is passed, the last checked time is still updated.
"""
now = datetime.now()
c = self.mysql_conn.cursor(MySQLdb.cursors.DictCursor)
select = "SELECT * FROM url_status WHERE url_hash=%s"
c.execute(select, urlkey)
r = c.fetchone()
if not r:
insert = "INSERT INTO url_status (url_hash, url, parent_hash, last_version, last_checked) VALUES (%s,%s,%s,%s,%s)"
c.execute(insert, (urlkey, url, parent_key, version, now))
else:
update = "UPDATE url_status SET last_version=%s, last_checked=%s WHERE url_hash=%s"
c.execute(update, (version, now, urlkey))
self.mysql_conn.commit()
last_version = r['last_version'] if r else None
if version and version != last_version:
if not c.execute("SELECT url_hash FROM url_history WHERE version=%s", version):
insert = "INSERT INTO url_history (url_hash, version, postdata_hash, created) VALUES (%s,%s,%s,%s)"
c.execute(insert, (urlkey, version, postdata_hash, now))
self.mysql_conn.commit()
def get_version_info(self, version):
"""Simple accessor method"""
c = self.mysql_conn.cursor(MySQLdb.cursors.DictCursor)
select = "SELECT * FROM url_history WHERE version=%s"
c.execute(select, version)
r = c.fetchone()
return (r['url_hash'], r['created']) if r else None

View File

@ -0,0 +1,87 @@
import sha
from datetime import datetime
from pydispatch import dispatcher
from scrapy.utils.misc import load_class
from scrapy.core import log, signals
from scrapy.core.exceptions import NotConfigured, IgnoreRequest
from scrapy.conf import settings
class HistoryMiddleware(object):
# How often we should re-check links we know about
MIN_CHECK_DAYS = 4
# How often we should process pages that have not changed (need to include depth)
MIN_PROCESS_UNCHANGED_DAYS = 12
def __init__(self):
historycls = load_class(settings['MEMORYSTORE'])
if not historycls:
raise NotConfigured
self.historydata = historycls()
dispatcher.connect(self.open_domain, signal=signals.domain_open)
dispatcher.connect(self.close_domain, signal=signals.domain_closed)
def process_request(self, request, spider):
key = urlkey(request.url)
status = self.historydata.status(domain, key)
if status:
_url, version, last_checked = status
d = datetime.now() - last_checked
if d.days < self.MIN_CHECK_DAYS:
raise IgnoreRequest("Not scraping %s (scraped %s ago)" % (request.url, d))
request.context['history_response_version'] = version
def process_response(self, request, response, spider):
version = request.context.get('history_response_version')
if version == response.version():
del request.content['history_response_version']
hist = self.historydata.version_info(domain, version)
if hist:
versionkey, created = hist
# if versionkey != urlkey(url) this means
# the same content is available on a different url
delta = datetime.now() - created
if delta.days < self.MIN_PROCESS_UNCHANGED_DAYS:
message = "skipping %s: unchanged for %s" % (response.url, delta)
raise IgnoreRequest(message)
self.record_visit(domain, request, response)
return response
def process_exception(self, request, exception, spider):
self.record_visit(spider.domain_name, request, None)
def open_domain(self, domain):
self.historydata.open(domain)
def close_domain(self, domain):
self.historydata.close_site(domain)
def record_visit(self, domain, request, response):
"""record the fact that the url has been visited"""
url = request.url
post_version = hash(request.body)
key = urlkey(url)
if response:
redirect_url = response.url
parentkey = urlkey(response.parent) if response.parent else None
version = response.version()
else:
redirect_url, parentkey, version = url, None, None
self.historydata.store(domain, key, url, parentkey, version, post_version)
def urlkey(url):
"""Generate a 'key' for a given url
>>> urlkey("http://www.example.com/")
'89e6a0649e06d83370cdf2cbfb05f363934a8d0c'
>>> urlkey("http://www.example.com/") == urlkey("http://www.example.com/?")
True
"""
from scrapy.utils.c14n import canonicalize
return hash(canonicalize(url))
def hash(value):
return sha.new(value).hexdigest() if value else None

View File

@ -0,0 +1,94 @@
from datetime import datetime
from twisted.internet import defer
from scrapy.core import log
from scrapy.core.scheduler import Scheduler
from scrapy.core.exceptions import IgnoreRequest
class RulesScheduler(Scheduler):
"""Scheduler that uses rules to determine if we should follow links
TODO:
* take into account where in chain of links we are (less depth should
be crawled more often)
* Be more strict about scraping product pages that rarely lead to new
versions of products. The same applies to pages with links. Particularly
useful for filtering out when there are many urls for the same product.
(but be careful to also filter out pages that almost always lead to new
output).
"""
# if these parameters change, then update bin/unavailable.py
# How often we should re-check links we know about
MIN_CHECK_DAYS = 4
# How often we should process pages that have not changed (need to include depth)
MIN_PROCESS_UNCHANGED_DAYS = 12
def enqueue_request(self, domain, request, priority=1):
"""Add a page to be scraped for a domain that is currently being scraped.
The url will only be added if we have not checked it already within
a specified time period.
"""
requestid = request.fingerprint()
added = self.groupfilter.add(domain, requestid)
if request.dont_filter or added:
key = urlkey(request.url) # we can not use fingerprint unless lost crawled history
status = self.historydata.status(domain, key)
now = datetime.now()
version = None
if status:
_url, version, last_checked = status
d = now - last_checked
if d.days < self.MIN_CHECK_DAYS:
log.msg("Not scraping %s (scraped %s ago)" % (request.url, d), level=log.DEBUG)
return
# put the version in the pending pages to avoid querying DB again
record = (request, version, now)
self.pending_requests[domain].put(record, priority)
def next_request(self, domain):
"""Get the next page from the superclass. This will add a callback
to prevent processing the page unless its content has been
changed.
In the event that it a page is not processed, the record_visit method
is called to update the last_checked time.
"""
pending_list = self.pending_requests.get(domain)
if not pending_list :
return None
request, version, timestamp = pending_list.get_nowait()[1]
post_version = hash(request.body)
def callback(pagedata):
"""process other callback if we pass the checks"""
if version == pagedata.version():
hist = self.historydata.version_info(domain, version)
if hist:
versionkey, created = hist
# if versionkey != urlkey(url) this means
# the same content is available on a different url
delta = timestamp - created
if delta.days < self.MIN_PROCESS_UNCHANGED_DAYS:
message = "skipping %s: unchanged for %s" % (pagedata.url, delta)
raise IgnoreRequest(message)
self.record_visit(domain, request.url, pagedata.url,
pagedata.parent, pagedata.version(),
post_version)
return pagedata
def errback(error) :
self.record_visit(domain, request.url, request.url, None, None,
post_version)
return error
d = defer.Deferred()
d.addCallbacks(callback, errback)
request.prepend_callback(d)
return request

View File

@ -0,0 +1,47 @@
"""
SQLHistoryStore
Persistent history storage using relational database
"""
from scrapy.contrib.history.history import URLHistory
from scrapy.core import log
from scrapy.conf import settings
class SQLHistoryStore(object) :
"""Implementation of a data store that stores information in a relation
database
This maintains a URLHistory object per site. That means each domain
has it's own session and is isolated from the others.
"""
def __init__(self):
self._store = {}
self._dbinfo = settings['SCRAPING_DB']
self._debug = settings['DEBUG_SQL_HISTORY_STORE']
def open(self, site):
self._store[site] = URLHistory(self._dbinfo)
def close_site(self, site):
self._store[site].close()
del self._store[site]
def store(self, site, key, url, parent=None, version=None, post_version=None):
history = self._store[site]
if self._debug:
log.msg("record_version(key=%s, url=%s, parent=%s, version=%s, post_version=%s)" %
(key, url, parent, version, post_version), domain=site, level=log.DEBUG)
history.record_version(key, url, parent, version, post_version)
def has_site(self, site):
return site in self._store
def status(self, site, key):
if site in self._store:
history = self._store[site]
return history.get_url_status(key)
def version_info(self, site, version):
history = self._store[site]
return history.get_version_info(version)

View File

@ -0,0 +1 @@
from scrapy.contrib.item.models import RobustScrapedItem, ValidationError, ValidationPipeline

View File

@ -0,0 +1,140 @@
"""
This module contains some extra base models for scraped items which could be
useful in some Scrapy implementations
"""
import sha
from scrapy.item import ScrapedItem
from scrapy.core.exceptions import UsageError, DropItem
class ValidationError(DropItem):
"""Indicates a data validation error"""
def __init__(self,problem,value=None):
self.problem = problem
self.value = value
def __str__(self):
if self.value is not None:
return '%s "%s"' % (self.problem, self.value)
else:
return '%s' % (self.problem)
class ValidationPipeline(object):
def process_item(self, domain, response, item):
item.validate()
return item
class RobustScrapedItem(ScrapedItem):
"""
A more robust scraped item class with a built-in validation mechanism and
minimal versioning support
"""
ATTRIBUTES = {
'guid': basestring, # a global unique identifier
'url': basestring, # the main URL where this item was scraped from
}
def __init__(self, data=None):
"""
A scraped item can be initialised with a dictionary that will be
squirted directly into the object.
"""
if isinstance(data, dict):
for attr, value in data.iteritems():
setattr(self, attr, value)
elif data is not None:
raise UsageError("Initialize with dict, not %s" % data.__class__.__name__)
self.__dict__['_version'] = None
def __getattr__(self, attr):
# Return None for valid attributes not set, raise AttributeError for invalid attributes
# Note that this method is called only when the attribute is not found in
# self.__dict__ or the class/instance methods.
if attr in self.ATTRIBUTES:
return None
else:
raise AttributeError(attr)
def __setattr__(self, attr, value):
"""
Set an attribute checking it matches the attribute type declared in self.ATTRIBUTES
"""
if not attr.startswith('_') and attr not in self.ATTRIBUTES:
raise AttributeError('Attribute "%s" is not a valid attribute name. You must add it to %s.ATTRIBUTES' % (attr, self.__class__.__name__))
if value is None:
self.__dict__.pop(attr, None)
return
type1 = self.ATTRIBUTES[attr]
if hasattr(type1, '__iter__'):
if not hasattr(value, '__iter__'):
raise TypeError('Attribute "%s" must be a sequence' % attr)
type2 = type1[0]
for i in value:
if not isinstance(i, type2):
raise TypeError('Attribute "%s" cannot contain %s, only %s' % (attr, i.__class__.__name__, type2.__name__))
else:
if not isinstance(value, type1):
raise TypeError('Attribute "%s" must be %s, not %s' % (attr, type1.__name__, value.__class__.__name__))
self.__dict__[attr] = value
self.__dict__['_version'] = None
def __delattr__(self, attr):
"""
Delete an attribute from the ScrapedItem instance if it exists.
If not, raise an AttributeError.
"""
if attr in self.__dict__:
del self.__dict__[attr]
self.__dict__['_version'] = None
else:
raise AttributeError("Attribute '%s' doesn't exist" % attr)
def __eq__(self, other):
return self.version == other.version
def __ne__(self, other):
return self.version != other.version
def __repr__(self):
# Generate this format so that it can be deserialized easily:
# ClassName({...})
reprdict = {}
for k, v in self.__dict__.iteritems():
if not k.startswith('_'):
reprdict[k] = v
return "%s(%s)" % (self.__class__.__name__, repr(reprdict))
def __str__(self) :
return "%s: GUID=%s, url=%s" % ( self.__class__.__name__ , self.guid, self.url )
def validate(self):
"""Method used to validate item attributes data"""
if not self.guid:
raise ValidationError('A guid is required')
def copy(self):
"""Create a new ScrapedItem object based on the current one"""
import copy
return copy.deepcopy(self)
@property
def version(self):
"""
Return a (cached) 40 char hash of all the item attributes.
WARNING: This cached version won't work if mutable products are
modified directly like:
item.features.append('feature')
"""
if self._version:
return self._version
hash_ = sha.new()
hash_.update("".join(["".join([n, str(v)]) for n,v in sorted(self.__dict__.iteritems())]))
return hash_.hexdigest()

View File

@ -0,0 +1,69 @@
import pprint
import gc
import socket
import libxml2
from pydispatch import dispatcher
from scrapy.core import signals
from scrapy.core.exceptions import NotConfigured
from scrapy.core.mail import MailSender
from scrapy.stats import stats
from scrapy.extension import extensions
from scrapy.conf import settings
class MemoryDebugger(object):
def __init__(self):
if not settings.getbool('MEMDEBUG_ENABLED'):
raise NotConfigured
self.mail = MailSender()
self.rcpts = settings.getlist('MEMDEBUG_NOTIFY')
self.domains_scraped = []
dispatcher.connect(self.domain_opened, signals.domain_opened)
dispatcher.connect(self.engine_started, signals.engine_started)
dispatcher.connect(self.engine_stopped, signals.engine_stopped)
def engine_started(self):
libxml2.debugMemory(1)
def engine_stopped(self):
figures = self.collect_figures()
report = self.create_report(figures)
self.print_or_send_report(report)
def collect_figures(self):
libxml2.cleanupParser()
gc.collect()
figures = []
if 'MemoryUsage' in extensions.enabled:
memusage = extensions.enabled['MemoryUsage']
memusage.update()
figures.append(("Memory usage at startup", int(memusage.data['startup']/1024/1024), "Mb"))
figures.append(("Maximum memory usage", int(memusage.data['max']/1024/1024), "Mb"))
figures.append(("Memory usage at shutdown", int(memusage.virtual/1024/1024), "Mb"))
figures.append(("Objects in gc.garbage", len(gc.garbage), ""))
figures.append(("libxml2 memory leak", libxml2.debugMemory(1), "bytes"))
return figures
def create_report(self, figures):
s = ""
s += "SCRAPY MEMORY DEBUGGER RESULTS\n\n"
for f in figures:
s += "%-30s : %s %s\n" % f
s += "\n"
if stats:
s += "SCRAPING STATS --------------------------------------------------\n\n"
s += pprint.pformat(stats)
return s
def print_or_send_report(self, report):
if self.rcpts:
self.mail.send(self.rcpts, "Scrapy Memory Debugger results at %s" % socket.gethostname(), report)
print report
def domain_opened(self, domain):
self.domains_scraped.append(domain)

View File

@ -0,0 +1,121 @@
import sys
import os
import pprint
import socket
from pydispatch import dispatcher
from scrapy.core import log, signals
from scrapy.core.manager import scrapymanager
from scrapy.core.engine import scrapyengine
from scrapy.core.exceptions import NotConfigured
from scrapy.core.mail import MailSender
from scrapy.stats import stats
from scrapy.conf import settings
class MemoryUsage(object):
_proc_status = '/proc/%d/status' % os.getpid()
_scale = {'kB': 1024.0, 'mB': 1024.0*1024.0,
'KB': 1024.0, 'MB': 1024.0*1024.0}
def __init__(self):
if not settings.getbool('MEMUSAGE_ENABLED'):
raise NotConfigured
if sys.platform != 'linux2':
raise NotConfigured("MemoryUsage extension is only available on Linux")
self.warned = False
self.data = {}
self.data['startup'] = 0
self.data['max'] = 0
scrapyengine.addtask(self.update, 60.0, now=True)
self.notify_mails = settings.getlist('MEMUSAGE_NOTIFY')
self.limit = settings.getint('MEMUSAGE_LIMIT_MB')*1024*1024
self.warning = settings.getint('MEMUSAGE_WARNING_MB')*1024*1024
self.report = settings.getbool('MEMUSAGE_REPORT')
if self.limit:
scrapyengine.addtask(self._check_limit, 60.0, now=True)
if self.warning:
scrapyengine.addtask(self._check_warning, 60.0, now=True)
self.mail = MailSender()
dispatcher.connect(self.engine_started, signal=signals.engine_started)
@property
def virtual(self):
return self._vmvalue('VmSize:')
@property
def resident(self):
return self._vmvalue('VmRSS:')
@property
def stacksize(self):
return self._vmvalue('VmStk:')
def engine_started(self):
self.data['startup'] = self.virtual
def update(self):
if self.virtual > self.data['max']:
self.data['max'] = self.virtual
def _vmvalue(self, VmKey):
# get pseudo file /proc/<pid>/status
try:
t = open(self._proc_status)
v = t.read()
t.close()
except:
return 0.0 # non-Linux?
# get VmKey line e.g. 'VmRSS: 9999 kB\n ...'
i = v.index(VmKey)
v = v[i:].split(None, 3) # whitespace
if len(v) < 3:
return 0.0 # invalid format?
# convert Vm value to bytes
return float(v[1]) * self._scale[v[2]]
def _check_limit(self):
if self.virtual > self.limit:
mem = self.limit/1024/1024
log.msg("Memory usage exceeded %dM. Shutting down Scrapy..." % mem, level=log.ERROR)
if self.notify_mails:
subj = "%s terminated: memory usage exceeded %dM at %s" % (settings['BOT_NAME'], mem, socket.gethostname())
self._send_report(self.notify_mails, subj)
scrapymanager.stop()
def _check_warning(self):
if self.warned: # warn only once
return
if self.virtual > self.warning:
mem = self.warning/1024/1024
log.msg("Memory usage reached %dM" % mem, level=log.WARNING)
if self.notify_mails:
subj = "%s warning: memory usage reached %dM at %s" % (settings['BOT_NAME'], mem, socket.gethostname())
self._send_report(self.notify_mails, subj)
self.warned = True
def _send_report(self, rcpts, subject):
"""send notification mail with some additional useful info"""
s = "Memory usage at engine startup : %dM\r\n" % (self.data['startup']/1024/1024)
s += "Maximum memory usage : %dM\r\n" % (self.data['max']/1024/1024)
s += "Current memory usage : %dM\r\n" % (self.virtual/1024/1024)
s += "ENGINE STATUS ------------------------------------------------------- \r\n"
s += "\r\n"
s += scrapyengine.getstatus()
s += "\r\n"
if stats:
s += "SCRAPING STATS ------------------------------------------------------ \r\n"
s += "\r\n"
s += pprint.pformat(stats)
self.mail.send(rcpts, subject, s)

View File

@ -0,0 +1,2 @@
from scrapy.contrib.pbcluster.worker.manager import ClusterWorker
from scrapy.contrib.pbcluster.master.web import ClusterMasterWeb

View File

@ -0,0 +1,220 @@
import urlparse
import urllib
import bisect
import sys
from pydispatch import dispatcher
from twisted.spread import pb
from twisted.internet import reactor
from twisted.python import util
from scrapy.core import log, signals
from scrapy.core.engine import scrapyengine
from scrapy.core.exceptions import NotConfigured
from scrapy.conf import settings
priorities = { 20:'NORMAL',
10:'QUICK',
0:'NOW',
}
# Set priorities module attributes
for val, attr in priorities.items():
setattr(sys.modules[__name__], "PRIORITY_%s" % attr, val )
class Node:
def __init__(self, remote, status, name):
self.__remote = remote
self._set_status(status)
self.name = name
def _set_status(self, status):
if not status:
self.available = False
else:
self.available = True
self.running = status['running']
self.pending = status['pending']
self.maxproc = status['maxproc']
self.starttime = status['starttime']
self.timestamp = status['timestamp']
self.loadavg = status['loadavg']
self.logdir = status['logdir']
self.lastcallresponse = status['callresponse']
def _remote_call(self, function, *args):
try:
deferred = self.__remote.callRemote(function, *args)
except pb.DeadReferenceError:
self._set_status(None)
log.msg("Lost connection to node %s." % (self.name), log.ERROR)
else:
deferred.addCallbacks(callback=self._set_status, errback=lambda reason: log.msg(reason, log.ERROR))
def get_status(self):
self._remote_call("status")
def schedule(self, domains, spider_settings=None, priority=PRIORITY_NORMAL):
self._remote_call("schedule", domains, spider_settings, priority)
def stop(self, domains):
self._remote_call("stop", domains)
def remove(self, domains):
self._remote_call("remove", domains)
class ClusterMaster(object):
def __init__(self):
if not settings.getbool('CLUSTER_MASTER_ENABLED'):
raise NotConfigured
self.nodes = {}
self.queue = []
dispatcher.connect(self._engine_started, signal=signals.engine_started)
def load_nodes(self):
def _make_callback(_factory, _name, _url):
def _errback(_reason):
log.msg("Could not get remote node %s in %s: %s." % (_name, _url, _reason), log.ERROR)
d = _factory.getRootObject()
d.addCallbacks(callback=lambda obj: self.add_node(obj, _name), errback=_errback)
"""Loads nodes from the CLUSTER_MASTER_NODES setting"""
for name, url in settings.get('CLUSTER_MASTER_NODES', {}).iteritems():
if name not in self.nodes:
server, port = url.split(":")
port = eval(port)
log.msg("Connecting to cluster worker %s..." % name)
log.msg("Server: %s, Port: %s" % (server, port))
factory = pb.PBClientFactory()
try:
reactor.connectTCP(server, port, factory)
except Exception, err:
log.msg("Could not connect to node %s in %s: %s." % (name, url, reason), log.ERROR)
else:
_make_callback(factory, name, url)
def update_nodes(self):
for node in self.nodes.itervalues():
node.get_status()
def add_node(self, cworker, name):
"""Add node given its node"""
node = Node(cworker, None, name)
node.get_status()
self.nodes[name] = node
log.msg("Added cluster worker %s" % name)
def remove_node(self, nodename):
raise NotImplemented
def schedule(self, domains, spider_settings=None, nodename=None, priority=PRIORITY_NORMAL):
if nodename:
self.nodes[nodename].schedule(domains, spider_settings, priority)
else:
self._dispatch_domains(domains, spider_settings, priority)
def stop(self, domains):
to_stop = {}
for domain in domains:
node = self.running.get(domain, None)
if node:
if node.name not in to_stop:
to_stop[node.name] = []
to_stop[node.name].append(domain)
for nodename, domains in to_stop.iteritems():
self.nodes[nodename].stop(domains)
def remove(self, domains):
to_remove = {}
for domain in domains:
node = self.pending.get(domain, None)
if node:
if node.name not in to_remove:
to_remove[node.name] = []
to_remove[node.name].append(domain)
for nodename, domains in to_remove.iteritems():
self.nodes[nodename].remove(domains)
def discard(self, domains):
"""Stop and remove all running and pending instances of the given
domains"""
self.remove(domains)
self.stop(domains)
@property
def running(self):
"""Return dict of running domains as domain -> node"""
d = {}
for node in self.nodes.itervalues():
for proc in node.running:
d[proc['domain']] = node
return d
@property
def pending(self):
"""Return dict of pending domains as domain -> node"""
d = {}
for node in self.nodes.itervalues():
for p in node.pending:
d[p['domain']] = node
return d
@property
def available_nodes(self):
return (node for node in self.nodes.itervalues() if node.available)
def _dispatch_domains(self, domains, spider_settings, priority):
"""Schedule the given domains in the availables nodes as good as
possible. The algorithm follows the next rules (in order):
1. search for nodes with available capacity(running < maxproc) and (if
any) schedules the domains there
2. if there isn't any node with available capacity it schedules the
domain in the node with the smallest number of pending spiders
"""
to_schedule = {} # domains to schedule per node
pending_node = [] # list of #pending, node
for node in self.available_nodes:
capacity = node.maxproc - len(node.running)
#order nodes in pending_node according to insertion position, calculated from priority comparison, for stage 2.
i = 0
for p in node.pending:
if p['priority'] <= priority:
i += 1
else:
break
bisect.insort(pending_node, (i, node))
#stage 1: use available capacity
to_schedule[node.name] = []
while domains and capacity > 0:
to_schedule[node.name].append(domains.pop(0))
capacity -= 1
if not domains:
break
#stage 2: queue in pendings the remaining domains.
# a) pops out minor insertion-point node b) schedules the domain c) reinserts the node in list with insertion-point incremented by one.
for domain in domains:
insert_point, node = pending_node.pop(0)
to_schedule[node.name].append(domain)
bisect.insort(pending_node, (insert_point+1, node))
for nodename, domains in to_schedule.iteritems():
if domains:
self.nodes[nodename].schedule(domains, spider_settings, priority)
def _engine_started(self):
self.load_nodes()
scrapyengine.addtask(self.update_nodes, settings.getint('CLUSTER_MASTER_POLL_INTERVAL'))

View File

@ -0,0 +1,174 @@
import datetime
from pydispatch import dispatcher
from scrapy.spider import spiders
from scrapy.management.web import banner, webconsole_discover_module
from scrapy.contrib.pbcluster.master.manager import ClusterMaster, priorities
class ClusterMasterWeb(ClusterMaster):
webconsole_id = 'cluster_master'
webconsole_name = 'Cluster master'
def __init__(self):
ClusterMaster.__init__(self)
dispatcher.connect(self.webconsole_discover_module, signal=webconsole_discover_module)
def webconsole_render(self, wc_request):
changes = ""
if wc_request.path == '/cluster_master/nodes/':
return self.render_nodes(wc_request)
elif wc_request.path == '/cluster_master/domains/':
return self.render_domains(wc_request)
elif wc_request.args:
changes = self.webconsole_control(wc_request)
s = self.render_header()
s += "<h2>Home</h2>\n"
s += "<table border='1'>\n"
s += "<tr><th>&nbsp;</th><th>Name</th><th>Available</th><th>Running</th><th>Pending</th><th>Load.avg</th></tr>\n"
for node in self.nodes.itervalues():
#chkbox = "<input type='checkbox' name='shutdown' value='%s' />" % domain if node.status in ["up", "idle"] else "&nbsp;"
nodelink = "<a href='nodes/#%s'>%s</a>" % (node.name, node.name)
chkbox = "&nbsp;"
loadavg = "%.2f %.2f %.2f" % node.loadavg
s += "<tr><td>%s</td><td>%s</td><td>%s</td><td>%d/%d</td><td>%d</td><td>%s</td></tr>\n" % \
(chkbox, nodelink, node.available, len(node.running), node.maxproc, len(node.pending), loadavg)
s += "</table>\n"
s += "</body>\n"
s += "</html>\n"
return str(s)
def webconsole_control(self, wc_request):
args = wc_request.args
if "updatenodes" in args:
self.update_nodes()
if "schedule" in args:
node = args["node"][0] if "node" in args else None
self.schedule(args["schedule"], nodename=node, priority=eval(args["priority"][0]))
if "stop" in args:
self.stop(args["stop"])
if "remove" in args:
self.remove(args["remove"])
return ""
def render_nodes(self, wc_request):
if wc_request.args:
self.webconsole_control(wc_request)
now = datetime.datetime.utcnow()
s = self.render_header()
for node in self.nodes.itervalues():
if node.available:
s += "<h2><a name='%s'>%s</h2>\n" % (node.name, node.name)
s += "<h3>Running domains</h3>\n"
if node.running:
s += "<form method='post' action='.'>\n"
s += "<table border='1'>\n"
s += "<tr><th>&nbsp;</th><th>PID</th><th>Domain</th><th>Status</th><th>Running time</th><th>Log file</th></tr>\n"
for proc in node.running:
chkbox = "<input type='checkbox' name='stop' value='%s' />" % proc['domain'] if proc['status'] == "running" else "&nbsp;"
start_time = proc.get('starttime', None)
elapsed = now - start_time if start_time else None
s += "<tr><td>%s</td><td>%s</td><td>%s</td><td>%s</td><td>%s</td><td>%s</td></tr>\n" % \
(chkbox, proc['pid'], proc['domain'], proc['status'], elapsed, proc['logfile'])
s += "</table>\n"
s += "<input type='hidden' name='node' value='%s'>\n" % node.name
s += "<p><input type='submit' value='Stop selected domains on %s'></p>\n" % node.name
s += "</form>\n"
else:
s += "<p>No running domains on %s</p>\n" % node.name
# pending domains
s += "<h3>Pending domains</h3>\n"
if node.pending:
s += "<form method='post' action='.'>\n"
s += "<select name='remove' multiple='multiple' size='10'>\n"
for p in node.pending:
s += "<option value='%s'>%s (P:%s)</option>\n" % (p['domain'], p['domain'],p['priority'])
s += "</select>\n"
s += "<input type='hidden' name='node' value='%s'>\n" % node.name
s += "<p><input type='submit' value='Remove selected pending domains on %s'></p>\n" % node.name
s += "</form>\n"
else:
s += "<p>No pending domains on %s</p>\n" % node.name
return str(s)
def render_domains(self, wc_request):
if wc_request.args:
self.webconsole_control(wc_request)
enabled_domains = set(spiders.asdict(include_disabled=False).keys())
print "Enabled domains: %s" % len(enabled_domains)
inactive_domains = enabled_domains - set(self.running.keys() + self.pending.keys())
s = self.render_header()
s += "<h2>Schedule domains</h2>\n"
s += "Inactive domains (not running or pending)<br />"
s += "<form method='post' action='.'>\n"
s += "<select name='schedule' multiple='multiple' size='10'>\n"
for domain in sorted(inactive_domains):
s += "<option>%s</option>\n" % domain
s += "</select>\n"
s += "</br>\n"
s += "Node (only available nodes shown):<br />\n"
s += "<select name='node'>\n"
s += "<option value='' selected='selected'>any</option>"
for node in self.available_nodes:
domcount = "%d/%d/%d" % (len(node.running), node.maxproc, len(node.pending))
loadavg = "%.2f %.2f %.2f" % node.loadavg
s += "<option value='%s'>%s [D: %s | LA: %s]</option>" % (node.name, node.name, domcount, loadavg)
s += "</select><br />\n"
s += "Priority:<br />\n"
s += "<select name='priority'>\n"
for p, pname in priorities.items():
if pname == "NORMAL":
s += "<option value='%s' selected>NORMAL</option>" % p
else:
s += "<option value='%s'>%s</option>" % (p, pname)
s += "</select>\n"
s += "<p><input type='submit' value='Schedule selected domains'></p>\n"
s += "</form>\n"
s += "<h2>Domains</h2>\n"
s += "<table border='1'>\n"
s += "<tr><th>Domain</th><th>Status</th><th>Node</th></tr>\n"
s += self._domains_table(self.running, '<b>running</b>')
s += self._domains_table(self.pending, 'pending')
s += "</table>\n"
return str(s)
def render_header(self):
s = banner(self)
s += "<p>Nav: "
s += "<a href='/cluster_master/'>Home</a> | "
s += "<a href='/cluster_master/domains/'>Domains</a> | "
s += "<a href='/cluster_master/nodes/'>Nodes</a> (<a href='/cluster_master/nodes/?updatenodes=1'>update</a>)"
s += "</p>"
return s
def _domains_table(self, dict_, status):
s = ""
for domain, node in dict_.iteritems():
s += "<tr><td>%s</td><td>%s</td><td>%s</td></tr>\n" % (domain, status, node.name)
return s
def webconsole_discover_module(self):
return self

View File

@ -0,0 +1,151 @@
import sys
import os
import time
import datetime
from twisted.internet import protocol, reactor
from twisted.spread import pb
from scrapy.core import log
from scrapy.core.exceptions import NotConfigured
from scrapy.conf import settings
from scrapy.core.engine import scrapyengine
class ScrapyProcessProtocol(protocol.ProcessProtocol):
def __init__(self, procman, domain, logfile=None, spider_settings=None):
self.procman = procman
self.domain = domain
self.logfile = logfile
self.start_time = datetime.datetime.utcnow()
self.status = "starting"
self.pid = -1
env = {'SCRAPY_LOGFILE': self.logfile, 'SCRAPY_CLUSTER_WORKER_ENABLED': '0', 'SCRAPY_WEBCONSOLE_ENABLED': '0'}
#We conserve original setting format for info purposes (avoid lots of unnecesary "SCRAPY_")
self.settings = spider_settings or {}
for k in self.settings:
env["SCRAPY_%s" % k] = self.settings[k]
self.env = env
def __str__(self):
return "<ScrapyProcess domain=%s, pid=%s, status=%s>" % (self.domain, self.pid, self.status)
def as_dict(self):
return {"domain": self.domain, "pid": self.pid, "status": self.status, "settings": self.settings, "logfile": self.logfile, "starttime": self.start_time}
def connectionMade(self):
self.pid = self.transport.pid
log.msg("ClusterWorker: started domain=%s, pid=%d, log=%s" % (self.domain, self.pid, self.logfile))
self.transport.closeStdin()
self.status = "running"
def processEnded(self, status_object):
log.msg("ClusterWorker: finished domain=%s, pid=%d, log=%s" % (self.domain, self.pid, self.logfile))
del self.procman.running[self.domain]
self.procman.next_pending()
class ClusterWorker(pb.Root):
def __init__(self):
if not settings.getbool('CLUSTER_WORKER_ENABLED'):
raise NotConfigured
self.maxproc = settings.getint('CLUSTER_WORKER_MAXPROC')
self.logdir = settings['CLUSTER_WORKER_LOGDIR']
self.running = {}
self.pending = []
self.starttime = time.time()
port = settings.getint('CLUSTER_WORKER_PORT')
scrapyengine.listenTCP(port, pb.PBServerFactory(self))
def remote_schedule(self, domains, spider_settings=None, priority=20):
"""Schedule new domains to be crawled in a separate processes"""
responses = []
for domain in domains:
if len(self.running) < self.maxproc and domain not in self.running:
self._run(domain, spider_settings)
responses.append("Started %s" % self.running[domain])
else:
i = 0
for p in self.pending:
if p['priority'] <= priority:
i += 1
else:
break
self.pending.insert(i, {'domain': domain, 'settings': spider_settings, 'priority': priority})
responses.append("Scheduled domain %s at position %s in queue" % (domain, i))
return self.status(responses)
def remote_stop(self, domains):
"""Stop running domains. For removing pending (not yet started) domains
use remove() instead"""
responses = []
for domain in domains:
if domain in self.running:
proc = self.running[domain]
log.msg("ClusterWorker: Sending shutdown signal to domain=%s, pid=%d" % (domain, proc.pid))
proc.transport.signalProcess('INT')
proc.status = "closing"
responses.append("Stopped process %s" % proc)
else:
responses.append("%s: domain not running." % domain)
return self.status(responses)
def remote_remove(self, domains):
"""Remove all scheduled instances of the given domains (if it hasn't
started yet). Otherwise use stop()"""
responses = []
for domain in domains:
to_remove = []
for p in self.pending:
if p['domain'] == domain:
to_remove.append(p)
for p in to_remove:
self.pending.remove(p)
responses.append("Unscheduled domain %s" % domain)
return self.status(responses)
def remote_status(self):
return self.status()
def status(self, response="Status Response"):
status = {}
status["pending"] = self.pending
status["running"] = [ self.running[k].as_dict() for k in self.running.keys() ]
status["starttime"] = self.starttime
status["timestamp"] = time.time()
status["maxproc"] = self.maxproc
status["loadavg"] = os.getloadavg()
status["logdir"] = self.logdir
status["callresponse"] = response
return status
def next_pending(self):
"""Run the next domain in the pending list, which is not already running"""
if len(self.running) >= self.maxproc:
return
for p in self.pending:
if p['domain'] not in self.running:
self._run(p['domain'], p['settings'])
self.pending.remove(p)
return
def _run(self, domain, spider_settings=None):
"""Spawn process to run the given domain. Don't call this method
directly. Instead use schedule()."""
logfile = os.path.join(self.logdir, domain, time.strftime("%FT%T.log"))
if not os.path.exists(os.path.dirname(logfile)):
os.makedirs(os.path.dirname(logfile))
scrapy_proc = ScrapyProcessProtocol(self, domain, logfile, spider_settings)
args = [sys.executable, sys.argv[0], 'crawl', domain]
proc = reactor.spawnProcess(scrapy_proc, sys.executable, args=args, env=scrapy_proc.env)
self.running[domain] = scrapy_proc

View File

@ -0,0 +1,29 @@
#!/usr/bin/python2.5
from twisted.spread import pb
from twisted.internet import reactor
from twisted.python import util
import sys
factory = pb.PBClientFactory()
reactor.connectTCP("localhost", 8789, factory)
d = factory.getRootObject()
sys.argv.pop(0)
if not sys.argv:
d.addCallback(lambda object: object.callRemote("status"))
elif sys.argv[0] == "-d":
try:
priority = eval(sys.argv[2])
except:
priority = 20
d.addCallback(lambda object: object.callRemote("schedule", [sys.argv[1]], priority=priority))
elif sys.argv[0] == "-s":
d.addCallback(lambda object: object.callRemote("stop", [sys.argv[1]]))
elif sys.argv[0] == "-r":
d.addCallback(lambda object: object.callRemote("remove", [sys.argv[1]]))
d.addCallbacks(callback = util.println, errback = lambda reason: 'error: '+str(reason.value))
d.addCallback(lambda _: reactor.stop())
reactor.run()

View File

@ -0,0 +1,37 @@
"""
A pipeline to persist objects using shove.
New is a "new generation" shelve. For more information see:
http://pypi.python.org/pypi/shove
"""
from string import Template
from shove import Shove
from pydispatch import dispatcher
from scrapy.core import signals
from scrapy.conf import settings
class ShoveItemPipeline(object):
def __init__(self):
self.uritpl = settings['SHOVEITEM_STORE_URI']
if not self.uritpl:
raise NotConfigured
self.opts = settings['SHOVEITEM_STORE_OPT'] or {}
self.stores = {}
dispatcher.connect(self.domain_open, signal=signals.domain_open)
dispatcher.connect(self.domain_closed, signal=signals.domain_closed)
def process_item(self, domain, response, item):
self.stores[domain][str(item.guid)] = item
return item
def domain_open(self, domain):
uri = Template(self.uritpl).substitute(domain=domain)
self.stores[domain] = Shove(uri, **self.opts)
def domain_closed(self, domain):
self.stores[domain].sync()

View File

@ -0,0 +1,15 @@
"""
Pipeline to print Items
"""
from scrapy.core import log
from scrapy.core.exceptions import NotConfigured
from scrapy.conf import settings
class ShowItemPipeline(object):
def __init__(self):
if not settings['DEBUG_SHOWITEM']:
raise NotConfigured
def process_item(self, domain, response, item):
log.msg("Scraped: \n%s" % repr(item), log.DEBUG, domain=domain)
return item

View File

@ -0,0 +1,34 @@
import time
from scrapy.core.exceptions import NotConfigured
from scrapy.store.db import DomainDataHistory
from scrapy.conf import settings
class LessScrapedPrioritizer(object):
"""
A spider prioritizer based on these few simple rules:
1. if spider was never scraped, it has top priority
2. if spider was scraped before, then the less recently the spider
has been scraped, the more priority it has
"""
def __init__(self, elements):
if not settings['SCRAPING_DB']:
raise NotConfigured("SCRAPING_DB setting is required")
self.ddh = DomainDataHistory(settings['SCRAPING_DB'], 'domain_data_history')
domains_to_scrape = set(elements)
self.priorities = {}
for domain in domains_to_scrape:
stat = self.ddh.getlast(domain, path="start_time")
if stat and stat[1]:
last_started = stat[1]
# spider is the timestamp of last start time
self.priorities[domain] = time.mktime(last_started.timetuple())
else:
# if domain was never scraped, it has top priority
self.priorities[domain] = 1
def get_priority(self, element):
return self.priorities[element]

View File

@ -0,0 +1,26 @@
"""
The ResponseSoup extension causes the Response objects to grow a new method
("getsoup") which returns a (cached) BeautifulSoup object of its body, and a
"soup" attribute with the same effect. The soup argument is provided for
convenience, but you cannot pass any BeautifulSoup constructor arguments (which
you can do with the getsoup() method).
For more information about BeautifulSoup see:
http://www.crummy.com/software/BeautifulSoup/documentation.html
"""
from BeautifulSoup import BeautifulSoup
from scrapy.http import Response
class ResponseSoup(object):
def __init__(self):
setattr(Response, 'getsoup', getsoup)
setattr(Response, 'soup', property(getsoup))
def getsoup(response, **kwargs):
if not hasattr(response, '_soup'):
body = response.body.to_string() if response.body is not None else ""
setattr(response, '_soup', BeautifulSoup(body, **kwargs))
return response._soup

View File

@ -0,0 +1,60 @@
"""
SpiderProfiler is an extension that hooks itself into every Request callback
returned from spiders to measure the processing time and memory allocation
caused by spiders code.
The results are collected using the StatsCollector.
This extension introduces a big impact on crawling performance, so enable only
when needed.
"""
import datetime
from pydispatch import dispatcher
from scrapy.extension import extensions
from scrapy.core import signals
from scrapy.core.exceptions import NotConfigured
from scrapy.stats import stats
from scrapy.conf import settings
class SpiderProfiler(object):
def __init__(self):
if not settings.getbool('SPIDERPROFILER_ENABLED'):
raise NotConfigured
dispatcher.connect(self._request_received, signals.request_received)
dispatcher.connect(self._engine_started, signals.engine_started)
def _engine_started(self):
self.memusage = extensions.enabled.get('MemoryUsage', None)
def _request_received(self, request, spider):
old_cbs = request.deferred.callbacks[0]
new_cbs = ((self._profiled_callback(old_cbs[0][0], spider), old_cbs[0][1], old_cbs[0][2]), old_cbs[1])
request.deferred.callbacks[0] = new_cbs
def _profiled_callback(self, function, spider):
def new_callback(*args, **kwargs):
tbefore = datetime.datetime.now()
mbefore = self._memusage()
r = function(*args, **kwargs)
tafter = datetime.datetime.now()
mafter = self._memusage()
ct = tafter-tbefore
tcc = stats.getpath('%s/profiling/total_callback_time' % spider.domain_name, datetime.timedelta(0))
sct = stats.getpath('%s/profiling/slowest_callback_time' % spider.domain_name, datetime.timedelta(0))
stats.setpath('%s/profiling/total_callback_time' % spider.domain_name, tcc+ct)
if ct > sct:
stats.setpath('%s/profiling/slowest_callback_time' % spider.domain_name, ct)
stats.setpath('%s/profiling/slowest_callback_name' % spider.domain_name, function.__name__)
stats.setpath('%s/profiling/slowest_callback_url' % spider.domain_name, args[0].url)
if self.memusage:
tma = stats.getpath('%s/profiling/total_mem_allocated_in_callbacks' % spider.domain_name, 0)
stats.setpath('%s/profiling/total_mem_allocated_in_callbacks' % spider.domain_name, tma+mafter-mbefore)
return r
return new_callback
def _memusage(self):
return self.memusage.virtual if self.memusage else 0.0

View File

@ -0,0 +1,18 @@
"""
Reload spider modules once they are finished scraping
This is to release any resources held on to by scraping spiders.
"""
import sys
from pydispatch import dispatcher
from scrapy.core import log, signals
class SpiderReloader(object):
def __init__(self):
dispatcher.connect(self.domain_closed, signal=signals.domain_closed)
def domain_closed(self, domain, spider):
module = spider.__module__
log.msg("reloading module %s" % module, domain=domain)
reload(sys.modules[module])

View File

@ -0,0 +1,36 @@
"""
DepthMiddleware is a scrape middleware used for tracking the depth of each
Request inside the site being scraped. It can be used to limit the maximum
depth to scrape or things like that
"""
from scrapy.core import log
from scrapy.http import Request
from scrapy.stats import stats
from scrapy.conf import settings
class DepthMiddleware(object):
def __init__(self):
self.maxdepth = settings.getint('DEPTH_LIMIT')
self.stats = settings.getbool('DEPTH_STATS')
if self.stats and self.maxdepth:
stats.setpath('_envinfo/request_depth_limit', self.maxdepth)
def process_result(self, response, result, spider):
def _filter(request):
if isinstance(request, Request):
request.depth = response.request.depth + 1
if self.maxdepth and request.depth > self.maxdepth:
log.msg("Ignoring link (depth > %d): %s " % (self.maxdepth, request.url), level=log.DEBUG, domain=spider.domain_name)
return False
elif self.stats:
stats.incpath('%s/request_depth_count/%s' % (spider.domain_name, request.depth))
if request.depth > stats.getpath('%s/request_depth_max' % spider.domain_name, 0):
stats.setpath('%s/request_depth_max' % spider.domain_name, request.depth)
return True
if self.stats and response.request.depth == 0: # otherwise we loose stats for depth=0
stats.incpath('%s/request_depth_count/0' % spider.domain_name)
return [r for r in result or () if _filter(r)]

View File

@ -0,0 +1,13 @@
from scrapy.core import log
from scrapy.http import Request
from scrapy.utils.url import url_is_from_spider
class OffsiteMiddleware(object):
def process_result(self, response, result, spider):
def _filter(r):
if isinstance(r, Request) and not url_is_from_spider(r.url, spider):
log.msg("Ignoring link (offsite): %s " % r.url, level=log.DEBUG, domain=spider.domain_name)
return False
return True
return [r for r in result or () if _filter(r)]

View File

@ -0,0 +1,12 @@
from scrapy.core import log
from scrapy.http import Request
class CrawlMiddleware(object):
def process_result(self, response, result, spider):
def _set_referer(r):
if isinstance(r, Request):
r.headers.setdefault('Referer', response.url)
return r
return [_set_referer(r) for r in result or ()]

View File

@ -0,0 +1,21 @@
"""
Scrapemiddlware to restrict crawling to only some particular URLs
"""
from scrapy.http import Request
from scrapy.core.exceptions import NotConfigured
from scrapy.conf import settings
class RestrictMiddleware(object):
def __init__(self):
self.allowed_urls = set(settings.getlist('RESTRICT_TO_URLS'))
if not self.allowed_urls:
raise NotConfigured
def process_result(self, response, result, spider):
def _filter(r):
if isinstance(r, Request) and r.url not in self.allowed_urls:
return False
return True
return [r for r in result or () if _filter(r)]

View File

@ -0,0 +1,22 @@
from scrapy.core import log
from scrapy.http import Request
from scrapy.core.exceptions import NotConfigured
from scrapy.conf import settings
class UrlLengthMiddleware(object):
"""This middleware discard requests with URLs longer than URLLENGTH_LIMIT"""
def __init__(self):
self.maxlength = settings.getint('URLLENGTH_LIMIT')
if not self.maxlength:
raise NotConfigured
def process_result(self, response, result, spider):
def _filter(request):
if isinstance(request, Request) and len(request.url) > self.maxlength:
log.msg("Ignoring link (url length > %d): %s " % (self.maxlength, request.url), level=log.DEBUG, domain=spider.domain_name)
return False
else:
return True
return [r for r in result or () if _filter(r)]

View File

@ -0,0 +1,115 @@
"""A django alike request-response model
most of this code is borrowed from django
"""
from Cookie import SimpleCookie
from scrapy.utils.datatypes import MultiValueDict, CaselessDict
def build_httprequest(twistedrequest):
"""Translate twisted request object to a django request approach"""
request = HttpRequest()
request.path = twistedrequest.path
request.method = twistedrequest.method.upper()
request.COOKIES = SimpleCookie(twistedrequest.received_cookies)
request.HEADERS = Headers(twistedrequest.received_headers)
request.ARGS = MultiValueDict(twistedrequest.args)
request.FILES = {} # not yet supported
request.content = twistedrequest.content
request.twistedrequest = twistedrequest
return request
class HttpRequest(object):
def __init__(self):
self.path = ''
self.method = None
self.COOKIES = {}
self.HEADERS = {}
self.ARGS = {}
self.FILES = {}
class HttpResponse(object):
status_code = 200
def __init__(self, content='', status=None, content_type=None):
content_type = content_type or "text/html; charset=utf-8"
self._headers = {'content-type': content_type}
self.content = content
self.cookies = SimpleCookie()
self.status_code = status
def __str__(self):
"Full HTTP message, including headers"
return '\n'.join(['%s: %s' % (key, value)
for key, value in self._headers.items()]) \
+ '\n\n' + self.content
def __setitem__(self, header, value):
self._headers[header.lower()] = value
def __delitem__(self, header):
try:
del self._headers[header.lower()]
except KeyError:
pass
def __getitem__(self, header):
return self._headers[header.lower()]
def has_header(self, header):
"Case-insensitive check for a header"
return self._headers.has_key(header.lower())
__contains__ = has_header
def items(self):
return self._headers.items()
def get(self, header, alternate):
return self._headers.get(header, alternate)
def set_cookie(self, key, value='', max_age=None, expires=None, path='/', domain=None, secure=None):
self.cookies[key] = value
for var in ('max_age', 'path', 'domain', 'secure', 'expires'):
val = locals()[var]
if val is not None:
self.cookies[key][var.replace('_', '-')] = val
def delete_cookie(self, key, path='/', domain=None):
self.cookies[key] = ''
if path is not None:
self.cookies[key]['path'] = path
if domain is not None:
self.cookies[key]['domain'] = domain
self.cookies[key]['expires'] = 0
self.cookies[key]['max-age'] = 0
class Headers(CaselessDict):
def __init__(self, source=None, encoding='utf-8'):
self.encoding = encoding
if getattr(source, 'iteritems', None):
d = source.iteritems()
else:
d = source # best effort
# can't use CaselessDict.__init__(self, d) because it doesn't call __setitem__
for k,v in d:
self.__setitem__(k.lower(), v)
def normkey(self, key):
return key.title() # 'Content-Type' styles headers
def __setitem__(self, key, value):
"""Headers must not be unicode"""
if isinstance(key, unicode):
key = key.encode(self.encoding)
if isinstance(value, unicode):
value = value.encode(self.encoding)
super(Headers, self).__setitem__(key, value)

View File

@ -0,0 +1,129 @@
from functools import wraps
from twisted.internet import defer
from scrapy.utils.misc import mustbe_deferred
from .http import HttpResponse
JSONCALLBACK_RE = '^[a-zA-Z][a-zA-Z_.-]*$'
JSON_CONTENT_TYPES = ('application/json',)
def serialize(obj):
global _serialize
if not _serialize:
_loadserializers()
return _serialize(obj)
def unserialize(obj):
global _unserialize
if not _unserialize:
_loadserializers()
return _unserialize(obj)
class JsonException(Exception):
pass
class JsonResponse(HttpResponse):
def __init__(self, content=None, callback=None, serialize=serialize, *args, **kwargs):
content = serialize(content)
if callback: # JSONP support
status, content = 200, '%s(%s)' % (callback, content)
kwargs.setdefault('content_type', 'application/x-javascript')
HttpResponse.__init__(self, content=content, *args, **kwargs)
class JsonResponseAccepted(JsonResponse):
status_code = 202
class JsonResponseNoContent(JsonResponse):
status_code = 204
class JsonResponseNotModified(JsonResponse):
status_code = 304
class JsonResponseBadRequest(JsonResponse):
status_code = 400
class JsonResponseUnauthorized(JsonResponse):
status_code = 401
class JsonResponseForbidden(JsonResponse):
status_code = 403
class JsonResponseNotFound(JsonResponse):
status_code = 404
class JsonResponseInternalServerError(JsonResponse):
status_code = 500
class JsonResponseNotImplemented(JsonResponse):
status_code = 501
def json(func):
""" Decorator to wrap a json prepared view and return a JsonResponse
if content-type is application/json, sets request.JSON to unserialized request body.
in case of unserialization failure, returns JsonResponseBadRequest()
if returned data from func is a dictionary, serialize it and returns JsonResponse()
"""
if not hasattr(func, '__call__'):
raise TypeError('The argument should be a callable')
@wraps(func)
def wrapper(request, *args, **kwargs):
json_callback = request.ARGS.get('callback') # JSONP support
request.method = method = _x_http_method_override(request)
request.content_type = ct = content_type(request)
request.JSON = None
if method in ('POST', 'PUT'):
if ct in JSON_CONTENT_TYPES:
body = request.content.read()
try:
request.JSON = unserialize(body)
except Exception, e:
return JsonResponseBadRequest('Invalid json: %s' % e )
def _onsuccess(response):
if not isinstance(response, HttpResponse):
return JsonResponse(response, json_callback) # best effort
return response
ret = mustbe_deferred(func, request, *args, **kwargs)
ret.addCallback(_onsuccess)
return ret
return wrapper
def content_type(request):
ct = request.HEADERS.get('content-type','')
return ct.split(';')[0].strip()
def _x_http_method_override(request):
""" support for X-Http-Method-Override hack
some clients does not support methods others than GET and POST, that clients
has a chance to set an extra header to indicate intended method.
"""
return request.HEADERS.get('x-http-method-override', request.method).upper()
## json serializers
_serialize = _unserialize = None
def _loadserializers():
global _serialize, _unserialize
try:
import cjson
_serialize = cjson.encode
_unserialize = cjson.decode
except ImportError:
try:
import simplejson
_serialize = simplejson.dumps
_unserialize = simplejson.loads
except ImportError:
assert 0, 'json serialization needs cjson or simplejson modules'

View File

@ -0,0 +1,90 @@
import re
import urllib
import simplejson
from sha import sha
from twisted.internet import defer
from scrapy.core.engine import scrapyengine
from scrapy.spider import spiders
from scrapy.http import Request
from scrapy.item import ScrapedItem
from scrapy.core.exceptions import NotConfigured
from scrapy.conf import settings
from scrapy.utils.misc import memoize
from lrucache import LRUCache
from .site import WebSite, WebResource
from .http import HttpResponse
from .json import JsonResponse
JSONCALLBACK_RE = '^[a-zA-Z][a-zA-Z_.-]*$'
CACHESIZE = settings.get('WS_CACHESIZE', 20)
def _urlhash(request):
h = sha()
for a in sorted(request.ARGS):
h.update(request.ARGS[a])
return h.hexdigest()
@memoize(cache=LRUCache(CACHESIZE), hash=_urlhash)
def url_to_guid(httprequest):
url = httprequest.ARGS.get('url')
if not url:
return HttpResponse('Bad Request', 400)
url = urllib.unquote(url)
jsoncb = httprequest.ARGS.get('callback')
if jsoncb and not re.match(JSONCALLBACK_RE, jsoncb):
return HttpResponse('Bad callback argument', 400)
def _response(guids=(), message=None):
content = {
'guids': list(guids),
'domain': getattr(spider, 'domain_name', None),
'message': message,
}
return JsonResponse(content=content, callback=jsoncb)
spider = spiders.fromurl(url)
if not spider:
return _response(message='No crawler found for site')
if httprequest.ARGS.get('dontcrawl'):
return _response()
def _on_error(_failure):
return _response(message='Error downloading url from site')
def _on_success(pagedata):
try:
items = spider.identify(pagedata)
except Exception, ex:
return _response(message='Error processing url')
guids = [i.guid for i in items if isinstance(i, ScrapedItem)]
return _response(guids=guids)
deferred = defer.Deferred().addCallbacks(_on_success, _on_error)
request = Request(url=url, callback=deferred, dont_filter=True)
schd = scrapyengine.schedule(request, spider)
schd.chainDeferred(deferred)
return deferred
urlmapping = (
('^ws/tools/url_to_guid/$', url_to_guid),
)
class UrlToGuidService(WebSite):
def __init__(self):
if not settings['WS_ENABLED']:
raise NotConfigured
port = settings.getint('WS_PORT') or 8088
timeout = settings.getint('WS_TIMEOUT') or 15 # seconds
resource = WebResource(urlmapping, timeout=timeout)
WebSite.__init__(self, port=port, resource=resource)

View File

@ -0,0 +1,133 @@
"""Twisted website object as django
################################################################################
## Simple Usage example:
from twisted.internet import reactor
from scrapy.contrib.web.http import WebSite, HttpResponse
def helloword(request):
return HttpResponse('Hello World!')
def hello(request, name):
return HttpResponse('Hello %s' % name)
urls = (
('^hello/(?P<name>\w+)/$', hello),
('^$', helloword),
)
resource = WebResource(urls)
site = WebSite(port=8081, resource=resource)
reactor.run()
# now go to http://localhost:8081/
################################################################################
## Complex usage example:
from twisted.internet import reactor, defer
from scrapy.contrib.web.http import WebSite, HttpResponse
def delayed(request):
def _callback(result):
return HttpResponse('Heavy task completed: %s' % result)
def _errback(_failure):
return HttpResponse('Internal Server Error: %s' % _failure, status=500)
def heavytask(_):
import random
assert random.randint(0,1), "Exception found processing request"
return _
d = defer.Deferred().addCallback(heavytask)
d.addCallbacks(_callback, _errback)
reactor.callLater(1, d.callback, "Well done")
return d
urls = (('^delayed/$', delayed),)
resource = WebResource(urls)
site = WebSite(port=8081, resource=resource)
reactor.run()
"""
import re
from twisted.web import server, resource
from twisted.internet import reactor
from scrapy.utils.misc import mustbe_deferred
from .http import HttpResponse, build_httprequest
def urlresolver(urls, path):
"""Simple path to view mapper"""
path = path.lstrip('/')
for pathre, view in urls:
m = re.search(pathre, path)
if m:
kwargs = m.groupdict()
args = () if kwargs else m.groups()
return view, args, kwargs
return None, (), {}
class WebSite(server.Site):
def __init__(self, port=None, *args, **kwargs):
server.Site.__init__(self, *args, **kwargs)
if port:
self.bind(port)
def bind(self, port):
from scrapy.core.engine import scrapyengine
scrapyengine.listenTCP(port, self)
class WebResource(resource.Resource):
"""Translate twisted web approach to django alike way"""
isLeaf = True
debug = True
def __init__(self, urls, timeout=3, urlresolver=urlresolver):
resource.Resource.__init__(self)
self.urlresolver = urlresolver
self.timeout = timeout
self.urls = urls
def render(self, twistedrequest):
httprequest = build_httprequest(twistedrequest)
def _send_response(response):
assert isinstance(response, HttpResponse), 'view should return a HttpResponse object'
twistedrequest.setResponseCode(response.status_code or 200)
for key, val in response.items():
twistedrequest.setHeader(key, response[key])
twistedrequest.write(response.content)
twistedrequest.finish()
def _on_error(_failure):
content = _failure.getTraceback() if self.debug else 'Internal Error'
response = HttpResponse(content=str(_failure), status=500)
return _send_response(response)
view, args, kwargs = self.urlresolver(self.urls, httprequest.path)
if not view:
response = HttpResponse(content='Not Found', status=404)
_send_response(response)
return server.NOT_DONE_YET
deferred = mustbe_deferred(view, httprequest, *args, **kwargs)
deferred.addCallback(_send_response)
deferred.addErrback(_on_error)
if not deferred.timeoutCall:
deferred.setTimeout(self.timeout)
return server.NOT_DONE_YET

View File

@ -0,0 +1,55 @@
from functools import partial
from scrapy.core.exceptions import NotConfigured
from scrapy.utils.serialization import serialize
from scrapy.conf import settings
from .site import WebSite, WebResource
from .json import JsonResponse
STATS_WSPORT = settings.getint('STATS_WSPORT', 8089)
STATS_WSTIMEOUT = settings.getint('STATS_WSTIMEOUT', 15)
jsonserialize = partial(serialize, format='json')
def stats(request):
service = request.ARGS.get('service', 'getstats')
if service == 'getstats':
domain = request.ARGS.get('domain')
count = request.ARGS.get('count', 1)
if not domain:
path = request.ARGS.get('path')
offset = request.ARGS.get('offset')
order = request.ARGS.get('order', 'domain')
olist = request.ARGS.get('olist', 'ASC')
from scrapy.store.db import DomainDataHistory
ddh = DomainDataHistory(settings['SCRAPING_DB'], 'domain_data_history')
if service == 'countdomains':
stats = [ddh.domain_count()]
elif domain:
stats = list(ddh.get(domain, count=int(count)))
else:
stats = list(ddh.getlast_alldomains(count, offset, order, olist, path=path))
content = {'data': stats if stats else "No stats found"}
return JsonResponse(content=content, serialize=jsonserialize)
urlmapping = (
('^ws/tools/stats/$', stats),
)
class StatsService(WebSite):
def __init__(self):
if not settings['WS_ENABLED']:
raise NotConfigured
if not settings['SCRAPING_DB']:
print "SCRAPING_DB setting is required for the stats web service"
raise NotConfigured
resource = WebResource(urlmapping, timeout=STATS_WSTIMEOUT)
WebSite.__init__(self, port=STATS_WSPORT, resource=resource)

View File

@ -0,0 +1,27 @@
"""
Scheduler information module for Scrapy webconsole
"""
from pydispatch import dispatcher
from scrapy.core.engine import scrapyengine
from scrapy.management.web import banner
class EngineStatus(object):
webconsole_id = 'enginestatus'
webconsole_name = 'Engine status'
def __init__(self):
from scrapy.management.web import webconsole_discover_module
dispatcher.connect(self.webconsole_discover_module, signal=webconsole_discover_module)
def webconsole_render(self, wc_request):
s = banner(self)
s += "<pre><code>\n"
s += scrapyengine.getstatus()
s += "</pre></code>\n"
s += "</body>\n"
s += "</html>\n"
return s
def webconsole_discover_module(self):
return self

View File

@ -0,0 +1,79 @@
"""
Live statistics extension
"""
from datetime import datetime
from pydispatch import dispatcher
from scrapy.core import signals
from scrapy.core.engine import scrapyengine
from scrapy.management.web import banner
class SpiderStats(object):
def __init__(self):
self.scraped = 0
self.crawled = 0
self.started = None
self.finished = None
class LiveStats(object):
webconsole_id = 'livestats'
webconsole_name = 'Spider live statistics of current run'
def __init__(self):
self.domains = {}
dispatcher.connect(self.domain_open, signal=signals.domain_open)
dispatcher.connect(self.domain_closed, signal=signals.domain_closed)
dispatcher.connect(self.item_scraped, signal=signals.item_scraped)
dispatcher.connect(self.response_downloaded, signal=signals.response_downloaded)
from scrapy.management.web import webconsole_discover_module
dispatcher.connect(self.webconsole_discover_module, signal=webconsole_discover_module)
def domain_open(self, domain, spider):
pstats = SpiderStats()
self.domains[spider.domain_name] = pstats
pstats.started = datetime.now()
pstats.finished = None
def domain_closed(self, domain, spider):
self.domains[spider.domain_name].finished = datetime.now()
def item_scraped(self, item, spider):
self.domains[spider.domain_name].scraped += 1
def response_downloaded(self, response, spider):
self.domains[spider.domain_name].crawled += 1
def webconsole_render(self, wc_request):
sch = scrapyengine.scheduler
dwl = scrapyengine.downloader
totdomains = totscraped = totcrawled = totscheduled = totactive = totpending = 0
s = banner(self)
s += "<table border='1'>\n"
s += "<tr><th>Domain</th><th>Items<br>Scraped</th><th>Pages<br>Crawled</th><th>Scheduler<br>Pending</th><th>Downloader<br/>Pending</th><th>Downloader<br/>Active</th><th>Start time</th><th>Finish time</th><th>Run time</th></tr>\n"
for d in sorted(self.domains.keys()):
scheduled = len(sch.pending_requests[d]) if d in sch.pending_requests else 0
active, pending = len(dwl.active_requests(d)), len(dwl.request_queue(d))
stats = self.domains[d]
runtime = stats.finished - stats.started if stats.finished else datetime.now() - stats.started
s += '<tr><td>%s</td><td align="right">%d</td><td align="right">%d</td><td align="right">%d</td><td align="right">%d</td><td align="right">%d</td><td>%s</td><td>%s</td><td>%s</td></tr>\n' % \
(d, stats.scraped, stats.crawled, scheduled, pending, active, str(stats.started), str(stats.finished), str(runtime))
totdomains += 1
totscraped += stats.scraped
totcrawled += stats.crawled
totscheduled += scheduled
totactive += active
totpending += pending
s += '<tr><td><b>%d domains</b></td><td align="right"><b>%d</b></td><td align="right"><b>%d</b></td><td align="right"><b>%d</b></td><td align="right"><b>%d</b></td><td align="right"><b>%d</b></td><td/><td/></tr>\n' % \
(totdomains, totscraped, totcrawled, totscheduled, totactive, totpending)
s += "</table>\n"
s += "</body>\n"
s += "</html>\n"
return s
def webconsole_discover_module(self):
return self

View File

@ -0,0 +1,41 @@
"""
Scheduler information module for Scrapy webconsole
"""
from pydispatch import dispatcher
from scrapy.core import signals
from scrapy.core.engine import scrapyengine
from scrapy.spider import spiders
from scrapy.management.web import banner
class SchedulerStats(object):
webconsole_id = 'scheduler'
webconsole_name = 'Scheduler queue'
def __init__(self):
from scrapy.management.web import webconsole_discover_module
dispatcher.connect(self.webconsole_discover_module, signal=webconsole_discover_module)
def webconsole_render(self, wc_request):
s = banner(self)
s += "<ul>\n"
for domain, requests in scrapyengine.scheduler.pending_requests.iteritems():
s += "<li>\n"
s += "%s (<b>%s</b> pages)\n" % (domain, len(requests))
s += "<ul>\n"
# requests is a tuple of request and deffered now, as I understand
for r, d in requests:
if hasattr(r, 'url'):
s += "<li><a href='%s'>%s</a></li>\n" % (r.url, r.url)
#else:
# s += "<li>%s</li>\n" % (repr(r))
s += "</ul>\n"
s += "</li>\n"
s += "</ul>\n"
s += "</body>\n"
s += "</html>\n"
return s
def webconsole_discover_module(self):
return self

View File

@ -0,0 +1,208 @@
"""
Extensions for allowing spider control from web console
"""
from pydispatch import dispatcher
from scrapy.core import signals
from scrapy.core.manager import scrapymanager
from scrapy.core.engine import scrapyengine
from scrapy.spider import spiders
from scrapy.management.web import banner
from scrapy.conf import settings
class Spiderctl(object):
webconsole_id = 'spiderctl'
webconsole_name = 'Spider control panel'
def __init__(self):
self.running = set()
self.finished = set()
dispatcher.connect(self.domain_open, signal=signals.domain_open)
dispatcher.connect(self.domain_closed, signal=signals.domain_closed)
from scrapy.management.web import webconsole_discover_module
dispatcher.connect(self.webconsole_discover_module, signal=webconsole_discover_module)
def domain_open(self, domain, spider):
self.running.add(domain)
def domain_closed(self, domain, spider):
if domain in self.running:
self.running.remove(domain)
self.finished.add(domain)
def webconsole_render(self, wc_request):
if wc_request.args:
changes = self.webconsole_control(wc_request)
enabled_domains = spiders.asdict(include_disabled=False).keys()
self.scheduled = set(scrapyengine.scheduler.pending_domains_count)
self.not_scheduled = [d for d in enabled_domains if d not in self.scheduled
and d not in self.running
and d not in self.finished]
s = banner(self)
s += '<p><a href="?reloadspiders=1">Reload spiders</a></p>\n'
s += '<table border=1">\n'
s += "<tr><th>Running (%d/%d)</th><th>Scheduled (%d)</th><th>Finished (%d)</th><th>Not scheduled (%d)</th></tr>\n" % \
(len(self.running),
settings['CONCURRENT_DOMAINS'],
len(self.scheduled),
len(self.finished),
len(self.not_scheduled))
s += "<tr>\n"
# running
s += "<td valign='top'>\n"
s += '<form method="post" action=".">\n'
s += '<select name="stop_running_domains" multiple="multiple">\n'
for domain in sorted(self.running):
s += "<option>%s</option>\n" % domain
s += '</select><br>\n'
s += '<br />'
s += '<input type="submit" value="Stop selected">\n'
s += '</form>\n'
s += "</td>\n"
# scheduled
s += "<td valign='top'>\n"
s += '<form method="post" action=".">\n'
s += '<select name="remove_pending_domains" multiple="multiple">\n'
for domain in sorted(self.scheduled):
s += "<option>%s</option>\n" % domain
s += '</select><br>\n'
s += '<br />'
s += '<input type="submit" value="Remove selected">\n'
s += '</form>\n'
s += "</td>\n"
# finished
s += "<td valign='top'>\n"
s += '<form method="post" action=".">\n'
s += '<select name="rerun_finished_domains" multiple="multiple">\n'
for domain in sorted(self.finished):
s += "<option>%s</option>\n" % domain
s += '</select><br>\n'
s += '<br />'
s += '<input type="submit" value="Re-schedule selected">\n'
s += '</form>\n'
s += "</td>\n"
# not scheduled
s += "<td valign='top'>\n"
s += '<form method="post" action=".">\n'
s += '<select name="add_pending_domains" multiple="multiple">\n'
for domain in sorted(self.not_scheduled):
s += "<option>%s</option>\n" % domain
s += '</select><br>\n'
s += '<br />'
s += '<input type="submit" value="Schedule selected">\n'
s += '</form>\n'
s += "</td>\n"
s += "</tr>\n"
s += "<tr>\n"
s += "<td>&nbsp;</td>\n"
s += "<td>\n"
s += '<form method="post" action=".">\n'
s += "<textarea name='bulk_remove_domains' rows='10' cols='25'></textarea>\n"
s += '<br>\n'
s += '<input type="submit" value="Bulk remove domains">\n'
s += '</form>\n'
s += '</td>\n'
s += "<td>&nbsp;</td>\n"
s += "<td>\n"
s += '<form method="post" action=".">\n'
s += "<textarea name='bulk_schedule_domains' rows='10' cols='25'></textarea>\n"
s += '<br>\n'
s += '<input type="submit" value="Bulk schedule domains">\n'
s += '</form>\n'
s += "</td>\n"
s += "</tr>\n"
s += "</table>\n"
if wc_request.args:
s += changes
s += "</body>\n"
s += "</html>\n"
return s
def webconsole_control(self, wc_request):
args = wc_request.args
s = "<hr />\n"
if "reloadspiders" in args:
scrapymanager.reload_spiders()
s += "<p><b>Spiders reloaded</b></p>\n"
return s
if "stop_running_domains" in args:
s += "<p>"
s += "Stopped spiders: <ul><li>%s</li></ul>" % "</li><li>".join(args["stop_running_domains"])
for domain in args["stop_running_domains"]:
scrapyengine.close_domain(domain)
s += "</p>"
if "remove_pending_domains" in args:
removed = []
for domain in args["remove_pending_domains"]:
if scrapyengine.scheduler.remove_pending_domain(domain):
removed.append(domain)
if removed:
s += "<p>"
s += "Removed scheduled spiders: <ul><li>%s</li></ul>" % "</li><li>".join(args["remove_pending_domains"])
s += "</p>"
if "bulk_remove_domains" in args:
scheduled = []
to_remove = set([d.strip() for d in args["bulk_remove_domains"][0].split("\n")])
removed = set([d for d in scrapyengine.scheduler.pending_domains if d in to_remove])
scrapyengine.scheduler.pending_domains = [d for d in execengine.scheduler.pending_domains if d not in removed]
if removed:
s += "<p>"
s += "Removed: <ul><li>%s</li></ul>" % "</li><li>".join(removed)
s += "</p>"
s += "<p>"
s += "Not removed: <ul><li>%s</li></ul>" % "</li><li>".join(to_remove - removed)
s += "</p>"
if "add_pending_domains" in args:
for domain in args["add_pending_domains"]:
if domain not in scrapyengine.scheduler.pending_domains_count:
scrapymanager.crawl(domain)
s += "<p>"
s += "Scheduled spiders: <ul><li>%s</li></ul>" % "</li><li>".join(args["add_pending_domains"])
s += "</p>"
if "bulk_schedule_domains" in args:
to_schedule = set([d.strip() for d in args["bulk_schedule_domains"][0].split("\n")])
scheduled = set()
for domain in to_schedule:
if domain in enabled_domains and domain not in scrapyengine.scheduler.pending_domains_count:
scrapymanager.crawl(domain)
scheduled.add(domain)
if scheduled:
s += "<p>"
s += "Scheduled: <ul><li>%s</li></ul>" % "</li><li>".join(scheduled)
s += "</p>"
s += "<p>"
s += "Not scheduled: <ul><li>%s</li></ul>" % "</li><li>".join(to_schedule - scheduled)
s += "</p>"
if "rerun_finished_domains" in args:
for domain in args["rerun_finished_domains"]:
if domain not in scrapyengine.scheduler.pending_domains_count:
scrapymanager.crawl(domain)
self.finished.remove(domain)
s += "<p>"
s += "Re-scheduled finished spiders: <ul><li>%s</li></ul>" % "</li><li>".join(args["rerun_finished_domains"])
s += "</p>"
return s
def webconsole_discover_module(self):
return self

View File

@ -0,0 +1,80 @@
"""
Extension for persistent statistics by domain
"""
import re
import pprint
from pydispatch import dispatcher
from scrapy.spider import spiders
from scrapy.core.exceptions import NotConfigured
from scrapy.management.web import banner
from scrapy.conf import settings
class SpiderStats(object):
webconsole_id = 'spiderstats'
webconsole_name = 'Spider stats (all-time)'
stats_mainpage = [
('item_scraped_count', "Items scraped <br>(last run)"),
('item_passed_count', "Items passed <br>(last run)"),
('response_count', "Pages crawled <br>(last run)"),
('start_time', "Start time <br>(last run)"),
('finish_time', "Finish time <br>(last run)"),
('finish_status', "Finish status<br>(last run)"),
]
PATH_RE = re.compile("/spiderstats/([^/]+)")
def __init__(self):
if not settings['SCRAPING_DB']:
raise NotConfigured("Requires SCRAPING_DB setting")
from scrapy.store.db import DomainDataHistory
self.ddh = DomainDataHistory(settings['SCRAPING_DB'], 'domain_data_history')
from scrapy.management.web import webconsole_discover_module
dispatcher.connect(self.webconsole_discover_module, signal=webconsole_discover_module)
def webconsole_render(self, wc_request):
m = self.PATH_RE.search(wc_request.path)
if m:
return self.render_domain(m.group(1))
else:
return self.render_main()
def render_main(self):
s = banner(self)
s += "<table border='1'>\n"
s += "<tr>\n"
s += "<th>Domain</th>"
for path, title in self.stats_mainpage:
s += "<th>%s</th>" % title
s += "</tr>\n"
for domain in spiders.asdict().keys():
s += "<tr>\n"
s += "<td><a href='%s'>%s</a></td>" % (domain, domain)
for path, title in self.stats_mainpage:
stat = self.ddh.getlast(domain, path=path)
value = stat[1] if stat else "None"
s += "<td>%s</td>" % value
s += "</tr>\n"
s += "</table>\n"
s += "</body>\n"
s += "</html>\n"
return str(s)
def render_domain(self, domain):
s = banner(self)
stats = self.ddh.getall(domain)
s += "<pre>\n"
s += pprint.pformat(list(stats))
s += "</pre>\n"
s += "</body>\n"
s += "</html>\n"
return str(s)
def webconsole_discover_module(self):
return self

View File

@ -0,0 +1,26 @@
import pprint
from pydispatch import dispatcher
from scrapy.stats import stats
from scrapy.management.web import banner, webconsole_discover_module
class StatsDump(object):
webconsole_id = 'stats'
webconsole_name = 'StatsCollector dump'
def __init__(self):
dispatcher.connect(self.webconsole_discover_module, signal=webconsole_discover_module)
def webconsole_render(self, wc_request):
s = banner(self)
s += "<pre><code>\n"
s += pprint.pformat(stats)
s += "</pre></code>\n"
s += "</body>\n"
s += "</html>\n"
return str(s)
def webconsole_discover_module(self):
return self

View File

@ -0,0 +1,3 @@
"""
Scrapy core library classes and functions.
"""

View File

@ -0,0 +1 @@
from scrapy.core.downloader.manager import Downloader

View File

@ -0,0 +1,79 @@
"""
Download handlers for different schemes
"""
from __future__ import with_statement
import os
import urlparse
from twisted.web.client import HTTPClientFactory
from twisted.internet import defer, reactor
from twisted.web import error as web_error
from scrapy.core import signals
from scrapy.http import Request, Response, Headers
from scrapy.core.exceptions import UsageError, HttpException
from scrapy.utils.misc import defer_succeed
from scrapy.conf import settings
def download_any(request, spider):
u = urlparse.urlparse(request.url)
if u.scheme == 'file':
return download_file(request, spider)
elif u.scheme in ('http', 'https'):
return download_http(request, spider)
else:
raise UsageError("Unsupported scheme '%s' in URL: <%s>" % (u.scheme, request.url))
def download_http(request, spider):
"""This functions handles http/https downloads"""
url = urlparse.urldefrag(request.url)[0]
agent = request.headers.get('user-agent', settings.get('USER_AGENT'))
request.headers.pop('user-agent', None) # remove user-agent if already present
factory = HTTPClientFactory(url=str(url), # never pass unicode urls to twisted
method=request.method,
postdata=request.body,
headers=request.headers,
agent=agent,
cookies=request.cookies,
timeout=getattr(spider, "download_timeout", None) or settings.getint('DOWNLOAD_TIMEOUT'),
followRedirect=False)
def _response(body):
body = body or ''
status = factory.status
parent = request.headers.get('Referer')
headers = Headers(factory.response_headers)
r = Response(domain=spider.domain_name, url=request.url, headers=headers, status=status, body=body, parent=parent)
signals.send_catch_log(signal=signals.request_uploaded, sender='download_http', request=request, spider=spider)
signals.send_catch_log(signal=signals.response_downloaded, sender='download_http', response=r, spider=spider)
return r
def _on_success(body):
return _response(body)
def _on_error(_failure):
ex = _failure.value
if isinstance(ex, web_error.Error): # HttpException
raise HttpException(ex.status, ex.message, _response(ex.response))
return _failure
factory.noisy = False
factory.deferred.addCallbacks(_on_success, _on_error)
u = urlparse.urlparse(request.url)
if u.scheme == 'https' :
from twisted.internet import ssl
contextFactory = ssl.ClientContextFactory()
reactor.connectSSL(u.hostname, u.port or 443, factory, contextFactory)
else:
reactor.connectTCP(u.hostname, u.port or 80, factory)
return factory.deferred
def download_file(request, spider) :
"""Return a deferred for a file download."""
filepath = request.url.split("file://")[1]
with open(filepath) as f:
response = Response(domain=spider.domain_name, url=request.url, body=f.read())
return defer_succeed(response)

Some files were not shown because too many files have changed in this diff Show More