urls

langroid/parsing/urls.py

url_to_tempfile(url)

Fetch content from the given URL and save it to a temporary local file.

Parameters:

Name	Type	Description	Default
url	str	The URL of the content to fetch.	required

Returns:

Name	Type	Description
str	str	The path to the temporary file where the content is saved.

Raises:

Type	Description
HTTPError	If there's any issue fetching the content.

Source code in langroid/parsing/urls.py

def url_to_tempfile(url: str) -> str:
    """
    Fetch content from the given URL and save it to a temporary local file.

    Args:
        url (str): The URL of the content to fetch.

    Returns:
        str: The path to the temporary file where the content is saved.

    Raises:
        HTTPError: If there's any issue fetching the content.
    """

    response = requests.get(url)
    response.raise_for_status()  # Raise an exception for HTTP errors

    # Create a temporary file and write the content
    with tempfile.NamedTemporaryFile(delete=False, suffix=".tmp") as temp_file:
        temp_file.write(response.content)
        return temp_file.name

get_user_input(msg, color='blue')

Prompt the user for input. Args: msg: printed prompt color: color of the prompt Returns: user input

Source code in langroid/parsing/urls.py

def get_user_input(msg: str, color: str = "blue") -> str:
    """
    Prompt the user for input.
    Args:
        msg: printed prompt
        color: color of the prompt
    Returns:
        user input
    """
    color_str = f"[{color}]{msg} " if color else msg + " "
    print(color_str, end="")
    return input("")

get_list_from_user(prompt="Enter input (type 'done' or hit return to finish)", n=None)

Prompt the user for inputs. Args: prompt: printed prompt n: how many inputs to prompt for. If None, then prompt until done, otherwise quit after n inputs. Returns: list of input strings

Source code in langroid/parsing/urls.py

def get_list_from_user(
    prompt: str = "Enter input (type 'done' or hit return to finish)",
    n: int | None = None,
) -> List[str]:
    """
    Prompt the user for inputs.
    Args:
        prompt: printed prompt
        n: how many inputs to prompt for. If None, then prompt until done, otherwise
            quit after n inputs.
    Returns:
        list of input strings
    """
    # Create an empty set to store the URLs.
    input_set = set()

    # Use a while loop to continuously ask the user for URLs.
    for _ in range(n or 1000):
        # Prompt the user for input.
        input_str = Prompt.ask(f"[blue]{prompt}")

        # Check if the user wants to exit the loop.
        if input_str.lower() == "done" or input_str == "":
            break

        # if it is a URL, ask how many to crawl
        if is_url(input_str):
            url = input_str
            input_str = Prompt.ask("[blue] How many new URLs to crawl?", default="0")
            max_urls = int(input_str) + 1
            tot_urls = list(find_urls(url, max_links=max_urls, max_depth=2))
            tot_urls_str = "\n".join(tot_urls)
            print(
                f"""
                Found these {len(tot_urls)} links upto depth 2:
                {tot_urls_str}
                """
            )

            input_set.update(tot_urls)
        else:
            input_set.add(input_str.strip())

    return list(input_set)

get_urls_paths_bytes_indices(inputs)

Given a list of inputs, return a list of indices of URLs, list of indices of paths, list of indices of byte-contents. Args: inputs: list of strings or bytes Returns: list of Indices of URLs, list of indices of paths, list of indices of byte-contents

Source code in langroid/parsing/urls.py

def get_urls_paths_bytes_indices(
    inputs: List[str | bytes],
) -> Tuple[List[int], List[int], List[int]]:
    """
    Given a list of inputs, return a
    list of indices of URLs, list of indices of paths, list of indices of byte-contents.
    Args:
        inputs: list of strings or bytes
    Returns:
        list of Indices of URLs,
        list of indices of paths,
        list of indices of byte-contents
    """
    urls = []
    paths = []
    byte_list = []
    for i, item in enumerate(inputs):
        if isinstance(item, bytes):
            byte_list.append(i)
            continue
        try:
            url_adapter = TypeAdapter(HttpUrl)
            Url(url=url_adapter.validate_python(item))
            urls.append(i)
        except ValidationError:
            if os.path.exists(item):
                paths.append(i)
            else:
                logger.warning(f"{item} is neither a URL nor a path.")
    return urls, paths, byte_list

crawl_url(url, max_urls=1)

Crawl starting at the url and return a list of URLs to be parsed, up to a maximum of max_urls. This has not been tested to work as intended. Ignore.

Source code in langroid/parsing/urls.py

def crawl_url(url: str, max_urls: int = 1) -> List[str]:
    """
    Crawl starting at the url and return a list of URLs to be parsed,
    up to a maximum of `max_urls`.
    This has not been tested to work as intended. Ignore.
    """
    from trafilatura.spider import focused_crawler

    if max_urls == 1:
        # no need to crawl, just return the original list
        return [url]

    to_visit = None
    known_urls = None

    # Create a RobotFileParser object
    robots = urllib.robotparser.RobotFileParser()
    while True:
        if known_urls is not None and len(known_urls) >= max_urls:
            break
        # Set the RobotFileParser object to the website's robots.txt file
        robots.set_url(url + "/robots.txt")
        robots.read()

        if robots.can_fetch("*", url):
            # Start or resume the crawl
            to_visit, known_urls = focused_crawler(
                url,
                max_seen_urls=max_urls,
                max_known_urls=max_urls,
                todo=to_visit,
                known_links=known_urls,
                rules=robots,
            )
        if to_visit is None:
            break

    if known_urls is None:
        return [url]
    final_urls = [s.strip() for s in known_urls]
    return list(final_urls)[:max_urls]

find_urls(url='https://en.wikipedia.org/wiki/Generative_pre-trained_transformer', max_links=20, visited=None, depth=0, max_depth=2, match_domain=True)

Recursively find all URLs on a given page.

Parameters:

Name	Type	Description	Default
url	str	The URL to start from.	'https://en.wikipedia.org/wiki/Generative_pre-trained_transformer'
max_links	int	The maximum number of links to find.	20
visited	set	A set of URLs that have already been visited.	None
depth	int	The current depth of the recursion.	0
max_depth	int	The maximum depth of the recursion.	2
match_domain	bool	Whether to restrict the crawl to url's own domain. When True (the default), only same-domain links are followed and returned. When False, links to other domains are followed as well, so the crawl will issue requests to hosts named by the pages it reads -- which may include untrusted external hosts, and internal or link-local addresses. find_urls does not consult robots.txt and does not delay between requests, so disable this only for domains you control or otherwise trust. max_links and max_depth still bound the crawl in both modes.	True

Returns:

Name	Type	Description
set	Set[str]	A set of URLs found on the page, at most max_links of them. Only http/https links that carry a host are followed and returned; links with any other scheme (e.g. mailto:, tel:, javascript:, ftp:) are skipped, including ones on the page's own domain, as are hostless references such as http:foo. The seed url is exempt from that filtering, though it is not guaranteed to survive the max_links truncation. These are URLs discovered, not necessarily URLs crawled: a page that alone supplies max_links links returns them without following any, so the result can contain links one hop beyond max_depth. max_depth bounds the requests issued, not the provenance of the returned set.

Source code in langroid/parsing/urls.py

def find_urls(
    url: str = "https://en.wikipedia.org/wiki/Generative_pre-trained_transformer",
    max_links: int = 20,
    visited: Optional[Set[str]] = None,
    depth: int = 0,
    max_depth: int = 2,
    match_domain: bool = True,
) -> Set[str]:
    """
    Recursively find all URLs on a given page.

    Args:
        url (str): The URL to start from.
        max_links (int): The maximum number of links to find.
        visited (set): A set of URLs that have already been visited.
        depth (int): The current depth of the recursion.
        max_depth (int): The maximum depth of the recursion.
        match_domain (bool): Whether to restrict the crawl to `url`'s own domain.
            When True (the default), only same-domain links are followed and
            returned. When False, links to other domains are followed as well,
            so the crawl will issue requests to hosts named by the pages it
            reads -- which may include untrusted external hosts, and internal
            or link-local addresses. `find_urls` does not consult robots.txt and
            does not delay between requests, so disable this only for domains
            you control or otherwise trust. `max_links` and `max_depth` still
            bound the crawl in both modes.

    Returns:
        set: A set of URLs found on the page, at most `max_links` of them. Only
            `http`/`https` links that carry a host are followed and returned;
            links with any other scheme (e.g. `mailto:`, `tel:`, `javascript:`,
            `ftp:`) are skipped, including ones on the page's own domain, as are
            hostless references such as `http:foo`. The seed `url` is exempt
            from that filtering, though it is not guaranteed to survive the
            `max_links` truncation.

            These are URLs *discovered*, not necessarily URLs crawled: a page
            that alone supplies `max_links` links returns them without
            following any, so the result can contain links one hop beyond
            `max_depth`. `max_depth` bounds the requests issued, not the
            provenance of the returned set.
    """

    if visited is None:
        visited = set()

    if url in visited or depth > max_depth:
        return visited

    visited.add(url)
    base_domain = urlparse(url).netloc

    try:
        response = requests.get(url, timeout=5)
        response.raise_for_status()
        soup = BeautifulSoup(response.text, "html.parser")
        links = [
            urljoin(response.url, a["href"])  # type: ignore
            for a in soup.find_all("a", href=True)
        ]

        # Defrag links: discard links that are to portions of same page
        defragged_links = list(
            set(urldefrag(link).url for link in links)  # type: ignore
        )

        # Keep only fetchable web links, then apply the domain filter if
        # requested.
        #
        # Both the scheme and host checks are load-bearing in both modes, and
        # neither is implied by the domain comparison once that comparison is
        # made conditional:
        #
        # - Schemes like `mailto:`, `javascript:`, `tel:` and `data:` parse to
        #   an empty netloc, so with `match_domain` True the domain comparison
        #   rejects them already -- but with it False that comparison is
        #   skipped, and such links would be returned and consume `max_links`.
        # - `ftp://`/`ws://` links to the page's own domain DO match
        #   `base_domain`, so the domain comparison admits them in either mode;
        #   previously they were returned even though `requests` cannot fetch
        #   them.
        # - Malformed references such as `http:foo` and `http:///path` have a
        #   web scheme but no host, so only the netloc check excludes them.
        #
        # Anything admitted here may be requested, and `find_urls` adds a URL
        # to `visited` before the request, so a junk entry survives the
        # exception the broad `except` below swallows and lands in the result.
        def is_fetchable_web_link(link: str) -> bool:
            parsed = urlparse(link)
            if parsed.scheme not in ("http", "https") or not parsed.netloc:
                return False
            return not match_domain or parsed.netloc == base_domain

        domain_matching_links = [
            link for link in defragged_links if is_fetchable_web_link(link)
        ]

        # Drop self-links under either identity of the page just fetched: the
        # requested `url` and, after a redirect, `response.url`. Links now
        # resolve against `response.url`, so a same-document anchor such as
        # `<a href="#section">` defrags to the redirect target; left in, it
        # would consume a `max_links` slot and be fetched a second time.
        # Ensure `url` is first, since below we take the first max_links urls.
        self_links = {url, urldefrag(response.url).url}
        domain_matching_links = [url] + [
            x for x in domain_matching_links if x not in self_links
        ]

        # If found links exceed max_links, return immediately
        if len(domain_matching_links) >= max_links:
            return set(domain_matching_links[:max_links])

        for link in domain_matching_links:
            if len(visited) >= max_links:
                break

            if link not in visited:
                visited.update(
                    find_urls(
                        link,
                        max_links,
                        visited,
                        depth + 1,
                        max_depth,
                        match_domain,
                    )
                )

    except (requests.RequestException, Exception) as e:
        print(f"Error fetching {url}. Error: {e}")

    return set(list(visited)[:max_links])