Spaces:
Runtime error
Runtime error
File size: 3,784 Bytes
105b369 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 |
from os import getenv
from typing import List, Optional
from phi.tools import Toolkit
from phi.utils.log import logger
try:
from apify_client import ApifyClient
except ImportError:
raise ImportError("`apify_client` not installed. Please install using `pip install apify-client`")
class ApifyTools(Toolkit):
def __init__(
self,
api_key: Optional[str] = None,
website_content_crawler: bool = True,
web_scraper: bool = False,
):
super().__init__(name="apify_tools")
self.api_key = api_key or getenv("MY_APIFY_TOKEN")
if not self.api_key:
logger.error("No Apify API key provided")
if website_content_crawler:
self.register(self.website_content_crawler)
if web_scraper:
self.register(self.web_scrapper)
def website_content_crawler(self, urls: List[str], timeout: Optional[int] = 60) -> str:
"""
Crawls a website using Apify's website-content-crawler actor.
:param urls: The URLs to crawl.
:param timeout: The timeout for the crawling.
:return: The results of the crawling.
"""
if self.api_key is None:
return "No API key provided"
if urls is None:
return "No URLs provided"
client = ApifyClient(self.api_key)
logger.debug(f"Crawling URLs: {urls}")
formatted_urls = [{"url": url} for url in urls]
run_input = {"startUrls": formatted_urls}
run = client.actor("apify/website-content-crawler").call(run_input=run_input, timeout_secs=timeout)
results: str = ""
for item in client.dataset(run["defaultDatasetId"]).iterate_items():
results += "Results for URL: " + item.get("url") + "\n"
results += item.get("text") + "\n"
return results
def web_scrapper(self, urls: List[str], timeout: Optional[int] = 60) -> str:
"""
Scrapes a website using Apify's web-scraper actor.
:param urls: The URLs to scrape.
:param timeout: The timeout for the scraping.
:return: The results of the scraping.
"""
if self.api_key is None:
return "No API key provided"
if urls is None:
return "No URLs provided"
client = ApifyClient(self.api_key)
logger.debug(f"Scrapping URLs: {urls}")
formatted_urls = [{"url": url} for url in urls]
page_function_string = """
async function pageFunction(context) {
const $ = context.jQuery;
const pageTitle = $('title').first().text();
const h1 = $('h1').first().text();
const first_h2 = $('h2').first().text();
const random_text_from_the_page = $('p').first().text();
context.log.info(`URL: ${context.request.url}, TITLE: ${pageTitle}`);
return {
url: context.request.url,
pageTitle,
h1,
first_h2,
random_text_from_the_page
};
}
"""
run_input = {
"pageFunction": page_function_string,
"startUrls": formatted_urls,
}
run = client.actor("apify/web-scraper").call(run_input=run_input, timeout_secs=timeout)
results: str = ""
for item in client.dataset(run["defaultDatasetId"]).iterate_items():
results += "Results for URL: " + item.get("url") + "\n"
results += item.get("pageTitle") + "\n"
results += item.get("h1") + "\n"
results += item.get("first_h2") + "\n"
results += item.get("random_text_from_the_page") + "\n"
return results
|