feat(assessment): auto comprehensive assessment - rules engine from college scoring standard, per-semester stats, manual overrides, word export
This commit is contained in:
root
@@ -0,0 +1,363 @@
|
||||
""" robotparser.py
|
||||
|
||||
Copyright (C) 2000 Bastian Kleineidam
|
||||
|
||||
You can choose between two licenses when using this package:
|
||||
1) GNU GPLv2
|
||||
2) PSF license for Python 2.2
|
||||
|
||||
The robots.txt Exclusion Protocol is implemented as specified in
|
||||
RFC 9309
|
||||
"""
|
||||
|
||||
import collections
|
||||
import re
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
|
||||
__all__ = ["RobotFileParser"]
|
||||
|
||||
RequestRate = collections.namedtuple("RequestRate", "requests seconds")
|
||||
|
||||
|
||||
class RobotFileParser:
|
||||
""" This class provides a set of methods to read, parse and answer
|
||||
questions about a single robots.txt file.
|
||||
|
||||
"""
|
||||
|
||||
def __init__(self, url=''):
|
||||
self.entries = []
|
||||
self.groups = {}
|
||||
self.sitemaps = []
|
||||
self.default_entry = None
|
||||
self.disallow_all = False
|
||||
self.allow_all = False
|
||||
self.set_url(url)
|
||||
self.last_checked = 0
|
||||
|
||||
def mtime(self):
|
||||
"""Returns the time the robots.txt file was last fetched.
|
||||
|
||||
This is useful for long-running web spiders that need to
|
||||
check for new robots.txt files periodically.
|
||||
|
||||
"""
|
||||
return self.last_checked
|
||||
|
||||
def modified(self):
|
||||
"""Sets the time the robots.txt file was last fetched to the
|
||||
current time.
|
||||
|
||||
"""
|
||||
import time
|
||||
self.last_checked = time.time()
|
||||
|
||||
def set_url(self, url):
|
||||
"""Sets the URL referring to a robots.txt file."""
|
||||
self.url = url
|
||||
self.host, self.path = urllib.parse.urlsplit(url)[1:3]
|
||||
|
||||
def read(self):
|
||||
"""Reads the robots.txt URL and feeds it to the parser."""
|
||||
try:
|
||||
f = urllib.request.urlopen(self.url)
|
||||
except urllib.error.HTTPError as err:
|
||||
if err.code in (401, 403):
|
||||
# If access to robot.txt has the status Unauthorized/Forbidden,
|
||||
# then most likely this applies to the entire site.
|
||||
self.disallow_all = True
|
||||
elif 400 <= err.code < 500:
|
||||
# RFC 9309, Section 2.3.1.3: the crawler MAY access any
|
||||
# resources on the server.
|
||||
self.allow_all = True
|
||||
elif 500 <= err.code < 600:
|
||||
# RFC 9309, Section 2.3.1.4: the crawler MUST assume
|
||||
# complete disallow.
|
||||
self.disallow_all = True
|
||||
err.close()
|
||||
else:
|
||||
raw = f.read()
|
||||
self.parse(raw.decode("utf-8", "surrogateescape").splitlines())
|
||||
|
||||
def _add_entry(self, entry):
|
||||
self.entries.append(entry)
|
||||
for agent in entry.useragents:
|
||||
agent = agent.lower()
|
||||
if agent not in self.groups:
|
||||
self.groups[agent] = entry
|
||||
else:
|
||||
self.groups[agent] = merge_entries(self.groups[agent], entry)
|
||||
|
||||
def parse(self, lines):
|
||||
"""Parse the input lines from a robots.txt file.
|
||||
|
||||
We allow that a user-agent: line is not preceded by
|
||||
one or more blank lines.
|
||||
"""
|
||||
entries = []
|
||||
# states:
|
||||
# 0: start state
|
||||
# 1: saw user-agent line
|
||||
# 2: saw an allow or disallow line
|
||||
state = 0
|
||||
entry = Entry()
|
||||
|
||||
self.modified()
|
||||
for line in lines:
|
||||
# remove optional comment and strip line
|
||||
i = line.find('#')
|
||||
if i >= 0:
|
||||
line = line[:i]
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
line = line.split(':', 1)
|
||||
if len(line) == 2:
|
||||
line[0] = line[0].strip().lower()
|
||||
line[1] = line[1].strip()
|
||||
if line[0] == "user-agent":
|
||||
if state == 2:
|
||||
self._add_entry(entry)
|
||||
entry = Entry()
|
||||
product_token = line[1]
|
||||
entry.useragents.append(product_token)
|
||||
state = 1
|
||||
elif line[0] == "disallow":
|
||||
if state != 0:
|
||||
state = 2
|
||||
try:
|
||||
entry.rulelines.append(RuleLine(line[1], False))
|
||||
except ValueError:
|
||||
pass
|
||||
elif line[0] == "allow":
|
||||
if state != 0:
|
||||
state = 2
|
||||
try:
|
||||
entry.rulelines.append(RuleLine(line[1], True))
|
||||
except ValueError:
|
||||
pass
|
||||
elif line[0] == "crawl-delay":
|
||||
if state != 0:
|
||||
# before trying to convert to int we need to make
|
||||
# sure that robots.txt has valid syntax otherwise
|
||||
# it will crash
|
||||
if line[1].strip().isdecimal():
|
||||
entry.delay = int(line[1])
|
||||
state = 2
|
||||
elif line[0] == "request-rate":
|
||||
if state != 0:
|
||||
numbers = line[1].split('/')
|
||||
# check if all values are sane
|
||||
if (len(numbers) == 2 and numbers[0].strip().isdecimal()
|
||||
and numbers[1].strip().isdecimal()):
|
||||
entry.req_rate = RequestRate(int(numbers[0]), int(numbers[1]))
|
||||
state = 2
|
||||
elif line[0] == "sitemap":
|
||||
# According to http://www.sitemaps.org/protocol.html
|
||||
# "This directive is independent of the user-agent line,
|
||||
# so it doesn't matter where you place it in your file."
|
||||
# Therefore we do not change the state of the parser.
|
||||
self.sitemaps.append(line[1])
|
||||
if state != 0:
|
||||
self._add_entry(entry)
|
||||
|
||||
def _find_entry(self, useragent):
|
||||
entry = self.groups.get(useragent.lower())
|
||||
if entry is not None:
|
||||
return entry
|
||||
for entry in self.groups.values():
|
||||
if entry.applies_to(useragent):
|
||||
return entry
|
||||
return self.groups.get('*')
|
||||
|
||||
def can_fetch(self, useragent, url):
|
||||
"""using the parsed robots.txt decide if useragent can fetch url"""
|
||||
if self.disallow_all:
|
||||
return False
|
||||
if self.allow_all:
|
||||
return True
|
||||
# Until the robots.txt file has been read or found not
|
||||
# to exist, we must assume that no url is allowable.
|
||||
# This prevents false positives when a user erroneously
|
||||
# calls can_fetch() before calling read().
|
||||
if not self.last_checked:
|
||||
return False
|
||||
# TODO: The private API is used in order to preserve an empty query.
|
||||
# This is temporary until the public API starts supporting this feature.
|
||||
parsed_url = urllib.parse._urlsplit(url, '')
|
||||
url = urllib.parse._urlunsplit(None, None, *parsed_url[2:])
|
||||
url = normalize_uri(url)
|
||||
if not url:
|
||||
url = "/"
|
||||
if url == '/robots.txt':
|
||||
# The /robots.txt URI is implicitly allowed.
|
||||
return True
|
||||
entry = self._find_entry(useragent)
|
||||
if entry is None:
|
||||
return True
|
||||
return entry.allowance(url)
|
||||
|
||||
def crawl_delay(self, useragent):
|
||||
if not self.mtime():
|
||||
return None
|
||||
entry = self._find_entry(useragent)
|
||||
if entry is None:
|
||||
return None
|
||||
return entry.delay
|
||||
|
||||
def request_rate(self, useragent):
|
||||
if not self.mtime():
|
||||
return None
|
||||
entry = self._find_entry(useragent)
|
||||
if entry is None:
|
||||
return None
|
||||
return entry.req_rate
|
||||
|
||||
def site_maps(self):
|
||||
if not self.sitemaps:
|
||||
return None
|
||||
return self.sitemaps
|
||||
|
||||
def __str__(self):
|
||||
entries = self.entries
|
||||
if self.default_entry is not None:
|
||||
entries = entries + [self.default_entry]
|
||||
return '\n\n'.join(filter(None, map(str, entries)))
|
||||
|
||||
class RuleLine:
|
||||
"""A rule line is a single "Allow:" (allowance==True) or "Disallow:"
|
||||
(allowance==False) followed by a path."""
|
||||
def __init__(self, path, allowance):
|
||||
if path == '' and not allowance:
|
||||
# an empty value means allow all
|
||||
allowance = True
|
||||
path = re.sub(r'[*]{2,}', '*', path)
|
||||
path = re.sub(r'[$][$*]+', '$', path)
|
||||
path = normalize_pattern(path)
|
||||
self.fullmatch = path.endswith('$')
|
||||
path = path.rstrip('$')
|
||||
if '$' in path:
|
||||
raise ValueError('$ not at the end of path')
|
||||
self.matcher = None
|
||||
if '*' in path:
|
||||
pattern = re.compile(translate_pattern(path), re.DOTALL)
|
||||
if self.fullmatch:
|
||||
self.matcher = pattern.fullmatch
|
||||
else:
|
||||
self.matcher = pattern.match
|
||||
self.path = path
|
||||
self.allowance = allowance
|
||||
|
||||
def applies_to(self, filename):
|
||||
# If the filename matches the rule, return the matching length plus 1.
|
||||
# If it does not match, return 0.
|
||||
if self.matcher is not None:
|
||||
m = self.matcher(filename)
|
||||
if m:
|
||||
return m.end() + 1
|
||||
else:
|
||||
if self.fullmatch:
|
||||
if filename == self.path:
|
||||
return len(self.path) + 1
|
||||
else:
|
||||
if filename.startswith(self.path):
|
||||
return len(self.path) + 1
|
||||
return 0
|
||||
|
||||
def __str__(self):
|
||||
return (("Allow" if self.allowance else "Disallow") + ": " + self.path
|
||||
+ ('$' if self.fullmatch else ''))
|
||||
|
||||
|
||||
class Entry:
|
||||
"""An entry has one or more user-agents and zero or more rulelines"""
|
||||
def __init__(self):
|
||||
self.useragents = []
|
||||
self.rulelines = []
|
||||
self.delay = None
|
||||
self.req_rate = None
|
||||
|
||||
def __str__(self):
|
||||
if not self.useragents:
|
||||
return ''
|
||||
ret = []
|
||||
for agent in self.useragents:
|
||||
ret.append(f"User-agent: {agent}")
|
||||
if self.delay is not None:
|
||||
ret.append(f"Crawl-delay: {self.delay}")
|
||||
if self.req_rate is not None:
|
||||
rate = self.req_rate
|
||||
ret.append(f"Request-rate: {rate.requests}/{rate.seconds}")
|
||||
if self.rulelines:
|
||||
ret.extend(map(str, self.rulelines))
|
||||
else:
|
||||
ret.append("Allow:")
|
||||
return '\n'.join(ret)
|
||||
|
||||
def applies_to(self, useragent):
|
||||
"""check if this entry applies to the specified agent"""
|
||||
if useragent is None:
|
||||
return '*' in self.useragents
|
||||
# split the name token and make it lower case
|
||||
useragent = useragent.split("/")[0].lower()
|
||||
for agent in self.useragents:
|
||||
if agent != '*':
|
||||
agent = agent.lower()
|
||||
if agent in useragent:
|
||||
return True
|
||||
return False
|
||||
|
||||
def allowance(self, filename):
|
||||
"""Preconditions:
|
||||
- our agent applies to this entry
|
||||
- filename is URL encoded
|
||||
"""
|
||||
best_match = -1
|
||||
allowance = True
|
||||
for line in self.rulelines:
|
||||
m = line.applies_to(filename)
|
||||
if m:
|
||||
if m > best_match:
|
||||
best_match = m
|
||||
allowance = line.allowance
|
||||
elif m == best_match and not allowance:
|
||||
allowance = line.allowance
|
||||
return allowance
|
||||
|
||||
|
||||
def normalize(path):
|
||||
unquoted = urllib.parse.unquote(path, errors='surrogateescape')
|
||||
return urllib.parse.quote(unquoted, errors='surrogateescape')
|
||||
|
||||
def normalize_uri(path):
|
||||
path, sep, query = path.partition('?')
|
||||
path = normalize(path)
|
||||
if sep:
|
||||
query = re.sub(r'[^=&]+', lambda m: normalize(m[0]), query)
|
||||
path += '?' + query
|
||||
return path
|
||||
|
||||
def normalize_pattern(path):
|
||||
path, sep, query = path.partition('?')
|
||||
path = re.sub(r'[^*$]+', lambda m: normalize(m[0]), path)
|
||||
if sep:
|
||||
query = re.sub(r'[^=&*$]+', lambda m: normalize(m[0]), query)
|
||||
path += '?' + query
|
||||
return path
|
||||
|
||||
def translate_pattern(path):
|
||||
parts = list(map(re.escape, path.split('*')))
|
||||
for i in range(1, len(parts)-1):
|
||||
parts[i] = f'(?>.*?{parts[i]})'
|
||||
parts[-1] = f'.*{parts[-1]}'
|
||||
return ''.join(parts)
|
||||
|
||||
def merge_entries(e1, e2):
|
||||
entry = Entry()
|
||||
entry.useragents = list(filter(set(e2.useragents).__contains__, e1.useragents))
|
||||
entry.rulelines = e1.rulelines + e2.rulelines
|
||||
entry.delay = e1.delay if e2.delay is None else e2.delay
|
||||
entry.req_rate = e1.req_rate if e2.req_rate is None else e2.req_rate
|
||||
return entry
|
||||
Reference in New Issue
Block a user