import os
import json
import time
from getpass import getpass
import pandas as pd
import requests
from IPython.display import display
def get_token():
"""Find the Apify token: the environment variable, then a hidden prompt."""
token = os.environ.get("APIFY_TOKEN")
if token:
return token.strip()
token = getpass("Paste your Apify API token (hidden): ")
return token.strip()
APIFY_TOKEN = get_token()
if APIFY_TOKEN:
print("Token found. It is kept in memory only and is never printed.")
else:
print("No token yet. Follow Step 1 above, then run this cell again.")
SCRAPER = "youtube-transcript-scraper" # a name from the table, any Cliqto Media scraper, or its link
INPUT = None # None = the example input below; or your own dict
MAX_ITEMS = 100
TIMEOUT_MINUTES = 10
# Tested example input for the 10 scrapers in the table. All of them are by Cliqto Media.
EXAMPLES = {
"youtube-transcript-scraper": {"videoUrls": ["jNQXAC9IVRw"], "preferredLanguages": ["en"]},
"reddit-scraper": {"mode": "subreddit", "subreddits": ["datascience"], "maxPosts": 10, "maxCommentsPerPost": 0},
"google-trends-scraper": {"searchTerms": ["web scraping"], "timeRange": "today 12-m"},
"google-autocomplete-scraper": {"keywords": ["python pandas"], "language": "en", "country": "us", "maxResults": 50},
"google-play-reviews-scraper": {"appIds": ["com.google.android.apps.translate"], "maxReviewsPerApp": 20, "maxTotalReviews": 20},
"trustpilot-reviews-scraper": {"companyDomains": ["apify.com"], "maxReviews": 20},
"pinterest-search-scraper": {"mode": "search", "queries": ["home office ideas"], "maxItems": 20},
"threads-scraper": {"mode": "profile", "usernames": ["zuck"], "maxItems": 20},
"website-contacts-scraper": {"startUrls": [{"url": "https://www.python.org/"}]},
"google-patents-scraper": {"publicationNumbers": ["US9014905B1"]},
}
OWNER = "cliqtomedia" # the notebook runs only Cliqto Media scrapers
def scraper_name(text):
"""Turn 'reddit-scraper', 'cliqtomedia/reddit-scraper' or a page link into 'reddit-scraper'."""
clean = str(text).strip().lower().split("?")[0].split("#")[0].replace("~", "/")
parts = [p for p in clean.split("/") if p]
if not parts:
raise ValueError("SCRAPER is empty. Enter a name from the table above.")
if len(parts) >= 2 and parts[-2] not in (OWNER, "actors"):
raise ValueError("This notebook runs only Cliqto Media scrapers (cliqtomedia/...). "
"%r is not one of them. See the full list: https://cliqtomedia.com/actors/" % str(text).strip())
return parts[-1]
def scraper_info(name):
"""Read the public page data of a Cliqto Media scraper from Apify. No token needed."""
try:
response = requests.get("https://api.apify.com/v2/acts/%s~%s" % (OWNER, name), timeout=60)
except requests.RequestException as err:
raise ValueError("Could not reach Apify. Check your internet connection and run this cell again. (%s)" % type(err).__name__)
if response.status_code == 404:
raise ValueError("There is no Cliqto Media scraper named %r. Check the spelling or see the full list: "
"https://cliqtomedia.com/actors/" % name)
if response.status_code >= 400:
raise ValueError("Apify returned an error (%s). Try again in a minute." % response.status_code)
info = response.json()["data"]
if info.get("username") != OWNER:
raise ValueError("This notebook runs only Cliqto Media scrapers.")
return info
def page_example(info, name):
"""The example input from the scraper's page, or the prefilled values of its input form."""
body = ((info.get("exampleRunInput") or {}).get("body") or "").strip()
try:
example = json.loads(body) if body else None
except ValueError:
example = None
if example and "helloWorld" not in example:
return example
try:
build = requests.get("https://api.apify.com/v2/acts/%s~%s/builds/default" % (OWNER, name), timeout=60).json()["data"]
schema = build.get("inputSchema") or "{}"
schema = json.loads(schema) if isinstance(schema, str) else schema
fields = schema.get("properties") or {}
return {key: field["prefill"] for key, field in fields.items() if "prefill" in field} or None
except (requests.RequestException, ValueError, KeyError, TypeError, AttributeError):
return None
ACTOR_ID, RUN_INPUT = None, None
try:
name = scraper_name(SCRAPER)
info = scraper_info(name)
if INPUT is not None:
RUN_INPUT, source = INPUT, "your INPUT"
elif name in EXAMPLES:
RUN_INPUT, source = EXAMPLES[name], "tested example"
else:
RUN_INPUT, source = page_example(info, name), "example from the scraper's page"
if not RUN_INPUT:
raise ValueError("This scraper has no example input. Put your own dict into INPUT. "
"The fields are on its page: https://apify.com/%s/%s" % (OWNER, name))
ACTOR_ID = OWNER + "/" + name
print("Scraper: %s (%s)" % (info.get("title") or name, ACTOR_ID))
print("Input (%s): %s" % (source, json.dumps(RUN_INPUT)))
print("Page with all input fields and the price: https://apify.com/%s/%s" % (OWNER, name))
if source != "tested example":
print("Check the price on that page before Step 3. Apify charges your account for the run.")
except ValueError as err:
print("Stopped:", err)
API = "https://api.apify.com/v2"
DONE = ("SUCCEEDED", "FAILED", "ABORTED", "TIMED-OUT")
class ApifyError(Exception):
"""An error with a message written for the notebook user."""
def api(method, path, token, **kwargs):
"""Call the Apify API and turn common errors into plain messages."""
try:
response = requests.request(
method,
API + path,
headers={"Authorization": "Bearer " + token},
timeout=120,
**kwargs,
)
except requests.RequestException as err:
raise ApifyError("Could not reach Apify. Check your internet connection and try again. (%s)" % type(err).__name__)
if response.status_code == 401:
raise ApifyError("Apify did not accept the token (401). Check that you copied the full Personal API token, then run Step 1 again.")
if response.status_code == 404:
raise ApifyError("Apify could not find that scraper (404). Check SCRAPER in Step 2 and run that cell again.")
if response.status_code == 403:
raise ApifyError("Your account cannot use this scraper (403). Open its page on Apify and check your plan.")
if response.status_code == 400:
raise ApifyError("Apify rejected the input (400). Check INPUT in Step 2 against the fields on the scraper's page.")
if response.status_code >= 400:
raise ApifyError("Apify returned an error (%s). Try again in a minute." % response.status_code)
return response.json()
def run_actor(actor_id, run_input, max_items, timeout_minutes, token, sleep=time.sleep):
"""Start a scraper, wait for it to finish, and return (run, items)."""
if not token:
raise ApifyError("No Apify token found. Follow Step 1 and run that cell again.")
if not actor_id:
raise ApifyError("No scraper chosen. Fix SCRAPER in Step 2 and run that cell again.")
actor = actor_id.strip().replace("/", "~")
try:
payload = json.loads(run_input) if isinstance(run_input, str) else run_input
except ValueError:
raise ApifyError("INPUT is not valid. Use a Python dict like the example printed in Step 2.")
started = time.time()
run = api("POST", "/acts/%s/runs" % actor, token, params={"waitForFinish": 60}, json=payload)["data"]
while run["status"] not in DONE:
if time.time() - started > timeout_minutes * 60:
raise ApifyError(
"The run is still going after %s minutes. It keeps running on Apify. "
"Open https://console.apify.com/actors/runs/%s to follow it." % (timeout_minutes, run["id"])
)
sleep(1)
run = api("GET", "/actor-runs/%s" % run["id"], token, params={"waitForFinish": 60})["data"]
if run["status"] in ("FAILED", "ABORTED", "TIMED-OUT"):
raise ApifyError(
"The run ended with status %s. Open https://console.apify.com/actors/runs/%s to read the log."
% (run["status"], run["id"])
)
items = api(
"GET",
"/datasets/%s/items" % run["defaultDatasetId"],
token,
params={"clean": "true", "format": "json", "limit": int(max_items)},
)
return run, items
items = None
try:
run, items = run_actor(ACTOR_ID, RUN_INPUT, MAX_ITEMS, TIMEOUT_MINUTES, APIFY_TOKEN)
print("Run %s finished with status %s. Items downloaded: %d." % (run["id"], run["status"], len(items)))
except ApifyError as err:
print("Stopped:", err)
df = None
if not items:
if items is not None:
print("The run finished but returned no items. Check INPUT in Step 2 and the run log on Apify.")
else:
print("No data yet. Fix the message above and run Step 3 again.")
else:
df = pd.json_normalize(items)
print("Rows: %d, columns: %d" % df.shape)
display(df.head())
print("Columns:")
print(list(df.columns))
def save_csv(frame, name="apify_results.csv"):
"""Write a CSV file. Lists and objects become JSON text so the file opens cleanly in a spreadsheet."""
out = frame.copy()
for column in out.columns:
out[column] = out[column].map(lambda v: json.dumps(v) if isinstance(v, (list, dict)) else v)
out.to_csv(name, index=False)
return name
if df is None or df.empty:
print("Nothing to save yet.")
else:
csv_name = save_csv(df)
print("Saved %s. Download it from the Files panel on the left." % csv_name)