courlan.urlstore

Domain-classified URL storage for web crawling workflows.

For usage examples, see the URL Store guide and the Web Crawling guide.

Defines a URL store which holds URLs along with relevant information and entails crawling helpers.

class courlan.urlstore.Compressor(compression=True)[source]

Bases: object

Use system information on available compression modules and define corresponding methods.

Parameters:

compression (bool)

compress(data)[source]

Pickle the data and compress it if a method is available.

Parameters:

data (Any)

Return type:

Any

compressor: Callable[[bytes], bytes]
decompress(data)[source]

Decompress the data if a method is available and load the object.

Parameters:

data (bytes)

Return type:

Any

decompressor: Callable[[bytes], bytes]
class courlan.urlstore.DomainEntry(state=State.OPEN)[source]

Bases: object

Class to record host-related information and URL paths.

Parameters:

state (State)

count: int
rules: bytes | RobotFileParser | None
state: State
timestamp: datetime | None
total: int
tuples: bytes | deque[UrlPathTuple]
class courlan.urlstore.State(*values)[source]

Bases: Enum

Record state information about a domain or host.

ALL_VISITED = 2
BUSTED = 3
OPEN = 1
class courlan.urlstore.UrlPathTuple(urlpath, visited)[source]

Bases: object

Class storing information for URL paths relative to a domain/host.

Parameters:
path()[source]

Get the URL path as string.

Return type:

str

urlpath: bytes
visited: bool
class courlan.urlstore.UrlStore(compressed=False, language=None, strict=False, trailing_slash=True, verbose=False)[source]

Bases: object

Defines a class to store domain-classified URLs and perform checks against it.

Thread-safety: readers are safe and writers are serialized, but a logical write is not globally atomic — drive mutations from a single writer thread.

Parameters:
add_from_html(htmlstring, url, external=False, lang=None, with_nav=True)[source]

Find links in a HTML document, filter them and add them to the data store.

Parameters:
Return type:

None

add_urls(urls=None, appendleft=None, visited=False)[source]

Add a list of URLs to the (possibly) existing one. Optional: append certain URLs to the left, specify if the URLs have already been visited.

Parameters:
Return type:

None

compressed: bool
discard(domains)[source]

Declare domains void and prune the store, http/https twins included.

Parameters:

domains (list[str])

Return type:

None

done: bool
download_threshold_reached(threshold)[source]

Find out if the download threshold (number of retrieved URLs) has been reached for one of the websites in store.

Parameters:

threshold (float)

Return type:

bool

dump_urls()[source]

Return a list of all known URLs.

Return type:

list[str]

establish_download_schedule(max_urls=100, time_limit=10)[source]

Get up to the specified number of URLs along with a suitable backoff schedule (in seconds).

Parameters:

  • max_urls (int)

  • time_limit (int)

Return type:

list[tuple[float, str]]

filter_unknown_urls(urls)[source]

Take a list of URLs and return the currently unknown ones.

Parameters:

urls (list[str])

Return type:

list[str]

filter_unvisited_urls(urls)[source]

Take a list of URLs and return the currently unvisited ones.

Parameters:

urls (list[str])

Return type:

list[str]

find_known_urls(domain)[source]

Get all already known URLs for the given domain (ex. “https://example.org”).

Parameters:

domain (str)

Return type:

list[str]

find_unvisited_urls(domain)[source]

Get all unvisited URLs for the given domain.

Parameters:

domain (str)

Return type:

list[str]

get_all_counts()[source]

Return all download counts for the hosts in store.

Return type:

list[int]

get_crawl_delay(website, default=5)[source]

Return the delay as extracted from robots.txt, or a given default.

Parameters:
Return type:

float

get_download_urls(time_limit=10.0, max_urls=10000)[source]

Get a list of immediately downloadable URLs according to the given time limit per domain.

Parameters:
Return type:

list[str]

get_known_domains()[source]

Return all known domains as a list.

Return type:

list[str]

get_rules(website)[source]

Return the stored crawling rules for the given website.

Parameters:

website (str)

Return type:

RobotFileParser | None

get_unvisited_domains()[source]

Find all domains for which there are unvisited URLs and potentially adjust done meta-information.

Return type:

list[str]

get_url(domain, as_visited=True)[source]

Retrieve a single URL and consider it to be visited (with corresponding timestamp).

Parameters:
Return type:

str | None

has_been_visited(url)[source]

Check if the given URL has already been visited.

Parameters:

url (str)

Return type:

bool

is_exhausted_domain(domain)[source]

Tell if all known URLs for the website have been visited.

Parameters:

domain (str)

Return type:

bool

is_known(url)[source]

Check if the given URL has already been stored.

Parameters:

url (str)

Return type:

bool

language: str | None
print_unvisited_urls()[source]

Print all unvisited URLs in store.

Return type:

None

print_urls()[source]

Print all URLs in store (URL + TAB + visited or not).

Return type:

None

reset()[source]

Re-initialize the URL store.

Return type:

None

store_rules(website, rules)[source]

Store crawling rules for a given website.

Parameters:
Return type:

None

strict: bool
total_url_number()[source]

Find number of all URLs in store.

Return type:

int

trailing_slash: bool
unvisited_websites_number()[source]

Return the number of websites for which there are still URLs to visit.

Return type:

int

urldict: defaultdict[str, DomainEntry]
write(filename)[source]

Write the URL store to disk as a pickle file, see load_store().

Parameters:

filename (str)

Return type:

None

courlan.urlstore.load_store(filename)[source]

Load a URL store from disk.

Warning: uses pickle, which can execute arbitrary code. Only load files you have written yourself with UrlStore.write().

Parameters:

filename (str)

Return type:

UrlStore