Need a proxy solution? Try ScrapeOps and get 1,000 free requests here, or compare all proxy providers here!
ScrapeOpsHow to Scrape Quora
Quora is a question and answer website that is home to a wealth of information. Since its founding in 2009, Quora has been a great place to ask a question and get an answer. This data could be used to understand consumer behavior, identify pain points, and discover new product opportunities.How to Scrape Quora With Requests and BeautifulSoup
In this tutorial, we'll learn how to scrape Quora using Python Requests and BeautifulSoup.- TLDR How to Scrape Quora
- How To Architect Our Scraper
- Understanding How To Scrape Quora
- Setting Up Our Quora Scraper
- Build A Quora Search Crawler
- Build A Quora Scraper
- Legal and Ethical Considerations
- Conclusion
- More Cool Articles
Need help scraping the web?
Then check out ScrapeOps, the complete toolkit for web scraping.
TLDR - How to Scrape Quora with Python
Need to scrape Quora but don't have time to code? Use the scraper below!Create a new project folder and add aconfig.json file with your "api_key". After you're done, add a file with the code below. Feel free to change any of the constants inside the main to tweak your results.import os
import csv
import requests
import json
import logging
from urllib.parse import urlencode
from bs4 import BeautifulSoup
import concurrent.futures
from dataclasses import dataclass, field, fields, asdict
API_KEY = ""
with open("config.json", "r") as config_file:
config = json.load(config_file)
API_KEY = config["api_key"]
def get_scrapeops_url(url, location="us"):
payload = {
"api_key": API_KEY,
"url": url,
"country": location,
"wait": 2000
}
proxy_url = "https://proxy.scrapeops.io/v1/?" + urlencode(payload)
return proxy_url
## Logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class SearchData:
name: str = ""
url: str = ""
rank: int = 0
def __post_init__(self):
self.check_string_fields()
def check_string_fields(self):
for field in fields(self):
# Check string fields
if isinstance(getattr(self, field.name), str):
# If empty set default text
if getattr(self, field.name) == "":
setattr(self, field.name, f"No {field.name}")
continue
# Strip any trailing spaces, etc.
value = getattr(self, field.name)
setattr(self, field.name, value.strip())
@dataclass
class ReplyData:
name: str = ""
reply: str = ""
def __post_init__(self):
self.check_string_fields()
def check_string_fields(self):
for field in fields(self):
# Check string fields
if isinstance(getattr(self, field.name), str):
# If empty set default text
if getattr(self, field.name) == "":
setattr(self, field.name, f"No {field.name}")
continue
# Strip any trailing spaces, etc.
value = getattr(self, field.name)
setattr(self, field.name, value.strip())
class DataPipeline:
def __init__(self, csv_filename="", storage_queue_limit=50):
self.names_seen = []
self.storage_queue = []
self.storage_queue_limit = storage_queue_limit
self.csv_filename = csv_filename
self.csv_file_open = False
def save_to_csv(self):
self.csv_file_open = True
data_to_save = []
data_to_save.extend(self.storage_queue)
self.storage_queue.clear()
if not data_to_save:
return
keys = [field.name for field in fields(data_to_save[0])]
file_exists = os.path.isfile(self.csv_filename) and os.path.getsize(self.csv_filename) > 0
with open(self.csv_filename, mode="a", newline="", encoding="utf-8") as output_file:
writer = csv.DictWriter(output_file, fieldnames=keys)
if not file_exists:
writer.writeheader()
for item in data_to_save:
writer.writerow(asdict(item))
self.csv_file_open = False
def is_duplicate(self, input_data):
if input_data.name in self.names_seen:
logger.warning(f"Duplicate item found: {input_data.name}. Item dropped.")
return True
self.names_seen.append(input_data.name)
return False
def add_data(self, scraped_data):
if self.is_duplicate(scraped_data) == False:
self.storage_queue.append(scraped_data)
if len(self.storage_queue) >= self.storage_queue_limit and self.csv_file_open == False:
self.save_to_csv()
def close_pipeline(self):
if self.csv_file_open:
time.sleep(3)
if len(self.storage_queue) > 0:
self.save_to_csv()
def scrape_search_results(keyword, location, page_number, data_pipeline=None, retries=3):
formatted_keyword = keyword.replace(" ", "+")
result_number = page_number * 10
url = f"https://www.google.com/search?q={formatted_keyword}%20site%3Aquora.com&start={result_number}"
tries = 0
success = False
while tries <= retries and not success:
try:
scrapeops_proxy_url = get_scrapeops_url(url, location=location)
response = requests.get(scrapeops_proxy_url)
logger.info(f"Recieved [{response.status_code}] from: {url}")
if response.status_code == 200:
success = True
else:
raise Exception(f"Failed request, Status Code {response.status_code}")
## Extract Data
soup = BeautifulSoup(response.text, "html.parser")
div_cards = soup.select("div span")
for div_card in div_cards:
name = div_card.find("h3")
link = div_card.find("a")
if not name or not link:
continue
result_number += 1
ranking = result_number
search_data = SearchData(
name=name.text,
url=link.get("href"),
rank=ranking
)
data_pipeline.add_data(search_data)
logger.info(f"Successfully parsed data from: {url}")
success = True
except Exception as e:
logger.error(f"An error occurred while processing page {url}: {e}")
logger.info(f"Retrying request for page: {url}, retries left {retries-tries}")
if not success:
raise Exception(f"Max Retries exceeded: {retries}")
def start_scrape(keyword, pages, location, data_pipeline=None, max_threads=5, retries=3):
with concurrent.futures.ThreadPoolExecutor(max_workers=max_threads) as executor:
executor.map(
scrape_search_results,
[keyword] * pages,
[location] * pages,
range(pages),
[data_pipeline] * pages,
[retries] * pages
)
def process_post(row, location, retries=3):
url = row["url"]
tries = 0
success = False
while tries <= retries and not success:
response = requests.get(get_scrapeops_url(url, location=location))
try:
if response.status_code == 200:
logger.info(f"Status: {response.status_code}")
soup = BeautifulSoup(response.text, "html.parser")
main_content = soup.select_one("div[id='mainContent']")
answer_cards = main_content.select("div div div div div[class='q-box']")
answer_pipeline = DataPipeline(csv_filename=f"{row['name'].replace(' ', '-')}.csv")
last_seen_name = ""
for answer_card in answer_cards:
excluded_words = ["All related", "Recommended"]
array = answer_card.text.split("·")
if len(array) < 3:
continue
promoted = "Promoted" in array[0]
related = "Related" in array[2][0:30] or "Related" in array[-2][0:30]
repeat_name = array[0] in last_seen_name or array[0] == last_seen_name
if promoted or related or repeat_name:
last_seen_name = array[0]
continue
reply_data = ReplyData(
name=array[0],
reply=array[-2]
)
answer_pipeline.add_data(reply_data)
answer_pipeline.close_pipeline()
success = True
else:
logger.warning(f"Failed Response: {response.status_code}")
raise Exception(f"Failed Request, status code: {response.status_code}")
except Exception as e:
logger.error(f"Exception thrown: {e}")
logger.warning(f"Failed to process page: {row['url']}")
logger.warning(f"Retries left: {retries-tries}")
tries += 1
if not success:
raise Exception(f"Max Retries exceeded: {retries}")
else:
logger.info(f"Successfully parsed: {row['url']}")
def process_results(csv_file, location, max_threads=5, retries=3):
logger.info(f"processing {csv_file}")
with open(csv_file, newline="") as file:
reader = list(csv.DictReader(file))
with concurrent.futures.ThreadPoolExecutor(max_workers=max_threads) as executor:
executor.map(
process_post,
reader,
[location] * len(reader),
[retries] * len(reader)
)
if __name__ == "__main__":
MAX_RETRIES = 3
MAX_THREADS = 5
PAGES = 5
LOCATION = "us"
logger.info(f"Crawl starting...")
## INPUT ---> List of keywords to scrape
keyword_list = ["learn rust"]
aggregate_files = []
## Job Processes
for keyword in keyword_list:
filename = keyword.replace(" ", "-")
crawl_pipeline = DataPipeline(csv_filename=f"{filename}.csv")
start_scrape(keyword, PAGES, LOCATION, data_pipeline=crawl_pipeline, max_threads=MAX_THREADS, retries=MAX_RETRIES)
crawl_pipeline.close_pipeline()
aggregate_files.append(f"{filename}.csv")
logger.info(f"Crawl complete.")
for file in aggregate_files:
process_results(file, LOCATION, max_threads=MAX_THREADS, retries=MAX_RETRIES)MAX_THREADS: Controls the number of threads that the program will use for concurrent execution.MAX_RETRIES: Defines the number of times the scraper will retry a failed request before giving up.PAGES: Determines how many pages of Google search results to scrape for each keyword.LOCATION: Specifies the geographical location (country) for the Google search.keyword_list: This is a list of keywords for which the script will perform the search and subsequent scraping.
How To Architect Our Quora Scraper
This project consists of two scrapers, a crawler, and a scraper. The job of the crawler is to perform a search on a certain topic, and the scraper will scrape individual posts from that search.Our crawler needs to perform the following actions:- Search Quora for a specific topic and parse the results.
- Use pagination to control our results.
- Save the extracted data with proper data storage.
- Concurrently search multiple pages of results at once.
- Use proxy integration to get past anti-bots any other potential roadblocks that may be in our way.
- Read the CSV from the crawl.
- Lookup each individual post from the CSV and parse the results.
- Store the relevant data from the step above.
- Perform the two steps above concurrently on multiple posts.
- Integrate with a proxy once again to avoid anti-bots and anything else that might get in our way.
Understanding How To Scrape Quora
Scraping Quora is not our typical project. Take a look at what happens when you try to search Quora without an account. You are immediately blocked and asked to login.As you saw above, we cannot actually perform a Quora search without logging in.However, there is a way around this. First, we need to get a better understanding of this task at a high level.As always, we're going to build both a crawler and a scraper.- The crawler will scrape Quora posts from Google.
- The scraper will pull information about those individual topics directly from Quora. If you're not familiar with scraping Google results, we've got an article on that here.
- Parsing
- Pagination
- Data Storage
- Concurrency
- Proxy Integration
Step 1: How To Request Quora Pages
To search Quora without an account, we're going to use Google. When you lookup Quora on Google, you are actually given an option to search Quora directly through Google.Take a look at the screenshot below.Here are the actual results from the search. If you look at the URL, this is something we can use to construct urls programmatically:https://www.google.com/search?q=learn%20rust%20site%3Aquora.com
https://www.google.com/search?q={formatted_keyword}%20site%3Aquora.com
https://www.quora.com/How-do-I-learn-the-Rust-programming-language
https://www.quora.com/Name-of-your-post
Step 2: How To Extract Data From Quora Results and Pages
Now, let's get a better look at the data we'll be extracting. Our Google results are nested incredibly deeply within the page. Our Quora results aren't much easier to extract because the replies are nested just as deeply.Here are the Google results. Here is an example Quora response card. Eachdiv card has a class of q-box.Step 3: How To Control Pagination
To control pagination with Google, we'll use the following layout in our URL:https://www.google.com/search?q={query}&start={page * 10}
- So page 0 would give us results 1 through 10.
- Page 2 gives us results 11 through 20 and so on and so forth.
Step 4: Geolocated Data
We don't need specific geolocation support for Quora, but we'll be using the ScrapeOps Proxy API which provides this for us anyway.When you use the API, you can give a parameter,country and you'll be routed through a server in the country of your choosing.- If we want to appear in the US, we can pass
"country": "us". - To appear in the UK, we can pass
"country": "uk".
Setting Up Our Quora Scraper Project
Let's get started. You can run the following commands to get setup.Create a New Project Foldermkdir quora-scraper cd quora-scraper
python -m venv venv
source venv/bin/activate
pip install requests
pip install beautifulsoup4
Build A Quora Search Crawler
As previously mentioned, we're going to use Google to crawl Quora. We'll add the following to our scraper in the next few sections:- Parsing
- Pagination
- Data Storage
- Concurrency
- Proxy Integration
Step 1: Create Simple Search Data Parser
We'll start by creating a parsing function. The code below sets our basic structure for the rest of the project. We begin with some retry logic and basic error handling but you really need to pay attention to our parsing logic.Here is the code we'll start with.import os
import csv
import requests
import json
import logging
from urllib.parse import urlencode
from bs4 import BeautifulSoup
import concurrent.futures
from dataclasses import dataclass, field, fields, asdict
API_KEY = ""
with open("config.json", "r") as config_file:
config = json.load(config_file)
API_KEY = config["api_key"]
def get_scrapeops_url(url, location="us"):
payload = {
"api_key": API_KEY,
"url": url,
"country": location,
"wait": 2000
}
proxy_url = "https://proxy.scrapeops.io/v1/?" + urlencode(payload)
return proxy_url
## Logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def scrape_search_results(keyword, location, retries=3):
formatted_keyword = keyword.replace(" ", "+")
url = f"https://www.google.com/search?q={formatted_keyword}%20site%3Aquora.com"
tries = 0
success = False
while tries <= retries and not success:
try:
scrapeops_proxy_url = get_scrapeops_url(url, location=location)
response = requests.get(scrapeops_proxy_url)
logger.info(f"Recieved [{response.status_code}] from: {url}")
if response.status_code == 200:
success = True
else:
raise Exception(f"Failed request, Status Code {response.status_code}")
## Extract Data
soup = BeautifulSoup(response.text, "html.parser")
div_cards = soup.select("div span")
for div_card in div_cards:
name = div_card.find("h3")
link = div_card.find("a")
if not name or not link:
continue
result_number += 1
ranking = result_number
search_data = {
"name": name.text,
"url": link.get("href"),
"rank": ranking
}
logger.info(f"Successfully parsed data from: {url}")
success = True
except Exception as e:
logger.error(f"An error occurred while processing page {url}: {e}")
logger.info(f"Retrying request for page: {url}, retries left {retries-tries}")
if not success:
raise Exception(f"Max Retries exceeded: {retries}")
if __name__ == "__main__":
MAX_RETRIES = 3
MAX_THREADS = 5
PAGES = 1
LOCATION = "us"
logger.info(f"Crawl starting...")
## INPUT ---> List of keywords to scrape
keyword_list = ["learn rust"]
aggregate_files = []
## Job Processes
for keyword in keyword_list:
filename = keyword.replace(" ", "-")
scrape_search_results(keyword, LOCATION, retries=MAX_RETRIES)
logger.info(f"Crawl complete.")- Find all of our result cards with
soup.select("div span") - For each
div_card:name = div_card.find("h3")finds the post name on Quora.link = div_card.find("a")finds the link to the post.
Step 2: Add Pagination
Adding pagination is a pretty simple task here. Google's results are all given a number and we get 10 results per page. Page 1 gives us results 1 through 10. Page 2 gives us results 11 through 20 and so on and so forth.To add these results, we need to add astart parameter to our URL.With the start param added in, our URL now looks like this:https://www.google.com/search?q={formatted_keyword}%20site%3Aquora.com&start={result_number}
start_scrape() function. This one is pretty simple at the moment. It iterates through our pages and runs scrape_search_results() on each page.def start_scrape(keyword, pages, location, data_pipeline=None, retries=3):
for page in range(pages):
scrape_search_results(keyword, location, page, data_pipeline=data_pipeline, retries=retries)import os
import csv
import requests
import json
import logging
from urllib.parse import urlencode
from bs4 import BeautifulSoup
import concurrent.futures
from dataclasses import dataclass, field, fields, asdict
API_KEY = ""
with open("config.json", "r") as config_file:
config = json.load(config_file)
API_KEY = config["api_key"]
def get_scrapeops_url(url, location="us"):
payload = {
"api_key": API_KEY,
"url": url,
"country": location,
"wait": 2000
}
proxy_url = "https://proxy.scrapeops.io/v1/?" + urlencode(payload)
return proxy_url
## Logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def scrape_search_results(keyword, location, page_number, retries=3):
formatted_keyword = keyword.replace(" ", "+")
result_number = page_number * 10
url = f"https://www.google.com/search?q={formatted_keyword}%20site%3Aquora.com&start={result_number}"
tries = 0
success = False
while tries <= retries and not success:
try:
scrapeops_proxy_url = get_scrapeops_url(url, location=location)
response = requests.get(scrapeops_proxy_url)
logger.info(f"Recieved [{response.status_code}] from: {url}")
if response.status_code == 200:
success = True
else:
raise Exception(f"Failed request, Status Code {response.status_code}")
## Extract Data
soup = BeautifulSoup(response.text, "html.parser")
div_cards = soup.select("div span")
for div_card in div_cards:
name = div_card.find("h3")
link = div_card.find("a")
if not name or not link:
continue
result_number += 1
ranking = result_number
search_data = {
"name": name.text,
"url": link.get("href"),
"rank": ranking
}
logger.info(f"Successfully parsed data from: {url}")
success = True
except Exception as e:
logger.error(f"An error occurred while processing page {url}: {e}")
logger.info(f"Retrying request for page: {url}, retries left {retries-tries}")
if not success:
raise Exception(f"Max Retries exceeded: {retries}")
def start_scrape(keyword, pages, location, data_pipeline=None, retries=3):
for page in range(pages):
scrape_search_results(keyword, location, page, data_pipeline=data_pipeline, retries=retries)
if __name__ == "__main__":
MAX_RETRIES = 3
MAX_THREADS = 5
PAGES = 1
LOCATION = "us"
logger.info(f"Crawl starting...")
## INPUT ---> List of keywords to scrape
keyword_list = ["learn rust"]
aggregate_files = []
## Job Processes
for keyword in keyword_list:
filename = keyword.replace(" ", "-")
start_scrape(keyword, PAGES, LOCATION, retries=MAX_RETRIES)
logger.info(f"Crawl complete.")Step 3: Storing the Scraped Data
Next, we need to store the data we've scraped. In order to do this, we need to add a couple classes. First, we'll add aSearchData class and then we'll add a DataPipeline class.SearchDatais adataclassthat exists specifically to hold data we've scraped. Once we've transformed a result on the page intoSearchData, we need to pass it into ourDataPipeline.- The
DataPipelinepipes this data to a CSV file and filters out our duplicates.
SearchData. It holds a name, url, and rank.@dataclass
class SearchData:
name: str = ""
url: str = ""
rank: int = 0
def __post_init__(self):
self.check_string_fields()
def check_string_fields(self):
for field in fields(self):
# Check string fields
if isinstance(getattr(self, field.name), str):
# If empty set default text
if getattr(self, field.name) == "":
setattr(self, field.name, f"No {field.name}")
continue
# Strip any trailing spaces, etc.
value = getattr(self, field.name)
setattr(self, field.name, value.strip())DataPipeline.class DataPipeline:
def __init__(self, csv_filename="", storage_queue_limit=50):
self.names_seen = []
self.storage_queue = []
self.storage_queue_limit = storage_queue_limit
self.csv_filename = csv_filename
self.csv_file_open = False
def save_to_csv(self):
self.csv_file_open = True
data_to_save = []
data_to_save.extend(self.storage_queue)
self.storage_queue.clear()
if not data_to_save:
return
keys = [field.name for field in fields(data_to_save[0])]
file_exists = os.path.isfile(self.csv_filename) and os.path.getsize(self.csv_filename) > 0
with open(self.csv_filename, mode="a", newline="", encoding="utf-8") as output_file:
writer = csv.DictWriter(output_file, fieldnames=keys)
if not file_exists:
writer.writeheader()
for item in data_to_save:
writer.writerow(asdict(item))
self.csv_file_open = False
def is_duplicate(self, input_data):
if input_data.name in self.names_seen:
logger.warning(f"Duplicate item found: {input_data.name}. Item dropped.")
return True
self.names_seen.append(input_data.name)
return False
def add_data(self, scraped_data):
if self.is_duplicate(scraped_data) == False:
self.storage_queue.append(scraped_data)
if len(self.storage_queue) >= self.storage_queue_limit and self.csv_file_open == False:
self.save_to_csv()
def close_pipeline(self):
if self.csv_file_open:
time.sleep(3)
if len(self.storage_queue) > 0:
self.save_to_csv()import os
import csv
import requests
import json
import logging
from urllib.parse import urlencode
from bs4 import BeautifulSoup
import concurrent.futures
from dataclasses import dataclass, field, fields, asdict
API_KEY = ""
with open("config.json", "r") as config_file:
config = json.load(config_file)
API_KEY = config["api_key"]
def get_scrapeops_url(url, location="us"):
payload = {
"api_key": API_KEY,
"url": url,
"country": location,
"wait": 2000
}
proxy_url = "https://proxy.scrapeops.io/v1/?" + urlencode(payload)
return proxy_url
## Logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class SearchData:
name: str = ""
url: str = ""
rank: int = 0
def __post_init__(self):
self.check_string_fields()
def check_string_fields(self):
for field in fields(self):
# Check string fields
if isinstance(getattr(self, field.name), str):
# If empty set default text
if getattr(self, field.name) == "":
setattr(self, field.name, f"No {field.name}")
continue
# Strip any trailing spaces, etc.
value = getattr(self, field.name)
setattr(self, field.name, value.strip())
class DataPipeline:
def __init__(self, csv_filename="", storage_queue_limit=50):
self.names_seen = []
self.storage_queue = []
self.storage_queue_limit = storage_queue_limit
self.csv_filename = csv_filename
self.csv_file_open = False
def save_to_csv(self):
self.csv_file_open = True
data_to_save = []
data_to_save.extend(self.storage_queue)
self.storage_queue.clear()
if not data_to_save:
return
keys = [field.name for field in fields(data_to_save[0])]
file_exists = os.path.isfile(self.csv_filename) and os.path.getsize(self.csv_filename) > 0
with open(self.csv_filename, mode="a", newline="", encoding="utf-8") as output_file:
writer = csv.DictWriter(output_file, fieldnames=keys)
if not file_exists:
writer.writeheader()
for item in data_to_save:
writer.writerow(asdict(item))
self.csv_file_open = False
def is_duplicate(self, input_data):
if input_data.name in self.names_seen:
logger.warning(f"Duplicate item found: {input_data.name}. Item dropped.")
return True
self.names_seen.append(input_data.name)
return False
def add_data(self, scraped_data):
if self.is_duplicate(scraped_data) == False:
self.storage_queue.append(scraped_data)
if len(self.storage_queue) >= self.storage_queue_limit and self.csv_file_open == False:
self.save_to_csv()
def close_pipeline(self):
if self.csv_file_open:
time.sleep(3)
if len(self.storage_queue) > 0:
self.save_to_csv()
def scrape_search_results(keyword, location, page_number, data_pipeline=None, retries=3):
formatted_keyword = keyword.replace(" ", "+")
result_number = page_number * 10
url = f"https://www.google.com/search?q={formatted_keyword}%20site%3Aquora.com&start={result_number}"
tries = 0
success = False
while tries <= retries and not success:
try:
scrapeops_proxy_url = get_scrapeops_url(url, location=location)
response = requests.get(scrapeops_proxy_url)
logger.info(f"Recieved [{response.status_code}] from: {url}")
if response.status_code == 200:
success = True
else:
raise Exception(f"Failed request, Status Code {response.status_code}")
## Extract Data
soup = BeautifulSoup(response.text, "html.parser")
div_cards = soup.select("div span")
for div_card in div_cards:
name = div_card.find("h3")
link = div_card.find("a")
if not name or not link:
continue
result_number += 1
ranking = result_number
search_data = SearchData(
name=name.text,
url=link.get("href"),
rank=ranking
)
data_pipeline.add_data(search_data)
logger.info(f"Successfully parsed data from: {url}")
success = True
except Exception as e:
logger.error(f"An error occurred while processing page {url}: {e}")
logger.info(f"Retrying request for page: {url}, retries left {retries-tries}")
if not success:
raise Exception(f"Max Retries exceeded: {retries}")
def start_scrape(keyword, pages, location, data_pipeline=None, retries=3):
for page in range(pages):
scrape_search_results(keyword, location, page, data_pipeline=data_pipeline, retries=retries)
if __name__ == "__main__":
MAX_RETRIES = 3
MAX_THREADS = 5
PAGES = 1
LOCATION = "us"
logger.info(f"Crawl starting...")
## INPUT ---> List of keywords to scrape
keyword_list = ["learn rust"]
aggregate_files = []
## Job Processes
for keyword in keyword_list:
filename = keyword.replace(" ", "-")
crawl_pipeline = DataPipeline(csv_filename=f"{filename}.csv")
start_scrape(keyword, PAGES, LOCATION, data_pipeline=crawl_pipeline, retries=MAX_RETRIES)
crawl_pipeline.close_pipeline()
aggregate_files.append(f"{filename}.csv")
logger.info(f"Crawl complete.")DataPipeline. This DataPipeline then gets passed into start_scrape() and then scrape_search_results(). We then use the add_data() method to add this data to our pipeline.Step 4: Adding Concurrency
Concurrency is vital when you're scraping at scale.start_scrape() already gives us the ability to run scrape_search_results() on multiple pages, but we want to run it on multiple pages simultaneously. In order to accomplish this, we'll be using ThreadPoolExecutor.Here is our new start_scrape() function.def start_scrape(keyword, pages, location, data_pipeline=None, max_threads=5, retries=3):
with concurrent.futures.ThreadPoolExecutor(max_workers=max_threads) as executor:
executor.map(
scrape_search_results,
[keyword] * pages,
[location] * pages,
range(pages),
[data_pipeline] * pages,
[retries] * pages
)scrape_search_resultsis the first argument passed intoexecutor.map(). This is the function we want to run on each available thread.- All other arguments get passed into
executor.map()as arrays that then get passed in on the individual threads.
import os
import csv
import requests
import json
import logging
from urllib.parse import urlencode
from bs4 import BeautifulSoup
import concurrent.futures
from dataclasses import dataclass, field, fields, asdict
API_KEY = ""
with open("config.json", "r") as config_file:
config = json.load(config_file)
API_KEY = config["api_key"]
def get_scrapeops_url(url, location="us"):
payload = {
"api_key": API_KEY,
"url": url,
"country": location,
"wait": 2000
}
proxy_url = "https://proxy.scrapeops.io/v1/?" + urlencode(payload)
return proxy_url
## Logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class SearchData:
name: str = ""
url: str = ""
rank: int = 0
def __post_init__(self):
self.check_string_fields()
def check_string_fields(self):
for field in fields(self):
# Check string fields
if isinstance(getattr(self, field.name), str):
# If empty set default text
if getattr(self, field.name) == "":
setattr(self, field.name, f"No {field.name}")
continue
# Strip any trailing spaces, etc.
value = getattr(self, field.name)
setattr(self, field.name, value.strip())
class DataPipeline:
def __init__(self, csv_filename="", storage_queue_limit=50):
self.names_seen = []
self.storage_queue = []
self.storage_queue_limit = storage_queue_limit
self.csv_filename = csv_filename
self.csv_file_open = False
def save_to_csv(self):
self.csv_file_open = True
data_to_save = []
data_to_save.extend(self.storage_queue)
self.storage_queue.clear()
if not data_to_save:
return
keys = [field.name for field in fields(data_to_save[0])]
file_exists = os.path.isfile(self.csv_filename) and os.path.getsize(self.csv_filename) > 0
with open(self.csv_filename, mode="a", newline="", encoding="utf-8") as output_file:
writer = csv.DictWriter(output_file, fieldnames=keys)
if not file_exists:
writer.writeheader()
for item in data_to_save:
writer.writerow(asdict(item))
self.csv_file_open = False
def is_duplicate(self, input_data):
if input_data.name in self.names_seen:
logger.warning(f"Duplicate item found: {input_data.name}. Item dropped.")
return True
self.names_seen.append(input_data.name)
return False
def add_data(self, scraped_data):
if self.is_duplicate(scraped_data) == False:
self.storage_queue.append(scraped_data)
if len(self.storage_queue) >= self.storage_queue_limit and self.csv_file_open == False:
self.save_to_csv()
def close_pipeline(self):
if self.csv_file_open:
time.sleep(3)
if len(self.storage_queue) > 0:
self.save_to_csv()
def scrape_search_results(keyword, location, page_number, data_pipeline=None, retries=3):
formatted_keyword = keyword.replace(" ", "+")
result_number = page_number * 10
url = f"https://www.google.com/search?q={formatted_keyword}%20site%3Aquora.com&start={result_number}"
tries = 0
success = False
while tries <= retries and not success:
try:
scrapeops_proxy_url = get_scrapeops_url(url, location=location)
response = requests.get(scrapeops_proxy_url)
logger.info(f"Recieved [{response.status_code}] from: {url}")
if response.status_code == 200:
success = True
else:
raise Exception(f"Failed request, Status Code {response.status_code}")
## Extract Data
soup = BeautifulSoup(response.text, "html.parser")
div_cards = soup.select("div span")
for div_card in div_cards:
name = div_card.find("h3")
link = div_card.find("a")
if not name or not link:
continue
result_number += 1
ranking = result_number
search_data = SearchData(
name=name.text,
url=link.get("href"),
rank=ranking
)
data_pipeline.add_data(search_data)
logger.info(f"Successfully parsed data from: {url}")
success = True
except Exception as e:
logger.error(f"An error occurred while processing page {url}: {e}")
logger.info(f"Retrying request for page: {url}, retries left {retries-tries}")
if not success:
raise Exception(f"Max Retries exceeded: {retries}")
def start_scrape(keyword, pages, location, data_pipeline=None, max_threads=5, retries=3):
with concurrent.futures.ThreadPoolExecutor(max_workers=max_threads) as executor:
executor.map(
scrape_search_results,
[keyword] * pages,
[location] * pages,
range(pages),
[data_pipeline] * pages,
[retries] * pages
)
if __name__ == "__main__":
MAX_RETRIES = 3
MAX_THREADS = 5
PAGES = 1
LOCATION = "us"
logger.info(f"Crawl starting...")
## INPUT ---> List of keywords to scrape
keyword_list = ["learn rust"]
aggregate_files = []
## Job Processes
for keyword in keyword_list:
filename = keyword.replace(" ", "-")
crawl_pipeline = DataPipeline(csv_filename=f"{filename}.csv")
start_scrape(keyword, PAGES, LOCATION, data_pipeline=crawl_pipeline, max_threads=MAX_THREADS, retries=MAX_RETRIES)
crawl_pipeline.close_pipeline()
aggregate_files.append(f"{filename}.csv")
logger.info(f"Crawl complete.")Step 5: Bypassing Anti-Bots
To bypass anti-bots, we'll make full use of the ScrapeOps Proxy API. This gets us past anti-bots and any other software that might be used to detect and block our scraper.We'll write a function,get_scrapeops_url() which takes in a url and a location. It also uses a wait parameter which tells ScrapeOps how long to wait before sending back our results.Our function takes in all of this information and then converts it to a proxied url with all of our custom parameters.def get_scrapeops_url(url, location="us"):
payload = {
"api_key": API_KEY,
"url": url,
"country": location,
"wait": 2000
}
proxy_url = "https://proxy.scrapeops.io/v1/?" + urlencode(payload)
return proxy_url"api_key"holds our ScrapeOps API key."url"represents the url we'd like to scrape."country"is the country we want to be routed through."wait"is the period we want ScrapeOps to wait before sending our results.
import os
import csv
import requests
import json
import logging
from urllib.parse import urlencode
from bs4 import BeautifulSoup
import concurrent.futures
from dataclasses import dataclass, field, fields, asdict
API_KEY = ""
with open("config.json", "r") as config_file:
config = json.load(config_file)
API_KEY = config["api_key"]
def get_scrapeops_url(url, location="us"):
payload = {
"api_key": API_KEY,
"url": url,
"country": location,
"wait": 2000
}
proxy_url = "https://proxy.scrapeops.io/v1/?" + urlencode(payload)
return proxy_url
## Logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class SearchData:
name: str = ""
url: str = ""
rank: int = 0
def __post_init__(self):
self.check_string_fields()
def check_string_fields(self):
for field in fields(self):
# Check string fields
if isinstance(getattr(self, field.name), str):
# If empty set default text
if getattr(self, field.name) == "":
setattr(self, field.name, f"No {field.name}")
continue
# Strip any trailing spaces, etc.
value = getattr(self, field.name)
setattr(self, field.name, value.strip())
class DataPipeline:
def __init__(self, csv_filename="", storage_queue_limit=50):
self.names_seen = []
self.storage_queue = []
self.storage_queue_limit = storage_queue_limit
self.csv_filename = csv_filename
self.csv_file_open = False
def save_to_csv(self):
self.csv_file_open = True
data_to_save = []
data_to_save.extend(self.storage_queue)
self.storage_queue.clear()
if not data_to_save:
return
keys = [field.name for field in fields(data_to_save[0])]
file_exists = os.path.isfile(self.csv_filename) and os.path.getsize(self.csv_filename) > 0
with open(self.csv_filename, mode="a", newline="", encoding="utf-8") as output_file:
writer = csv.DictWriter(output_file, fieldnames=keys)
if not file_exists:
writer.writeheader()
for item in data_to_save:
writer.writerow(asdict(item))
self.csv_file_open = False
def is_duplicate(self, input_data):
if input_data.name in self.names_seen:
logger.warning(f"Duplicate item found: {input_data.name}. Item dropped.")
return True
self.names_seen.append(input_data.name)
return False
def add_data(self, scraped_data):
if self.is_duplicate(scraped_data) == False:
self.storage_queue.append(scraped_data)
if len(self.storage_queue) >= self.storage_queue_limit and self.csv_file_open == False:
self.save_to_csv()
def close_pipeline(self):
if self.csv_file_open:
time.sleep(3)
if len(self.storage_queue) > 0:
self.save_to_csv()
def scrape_search_results(keyword, location, page_number, data_pipeline=None, retries=3):
formatted_keyword = keyword.replace(" ", "+")
result_number = page_number * 10
url = f"https://www.google.com/search?q={formatted_keyword}%20site%3Aquora.com&start={result_number}"
tries = 0
success = False
while tries <= retries and not success:
try:
scrapeops_proxy_url = get_scrapeops_url(url, location=location)
response = requests.get(scrapeops_proxy_url)
logger.info(f"Recieved [{response.status_code}] from: {url}")
if response.status_code == 200:
success = True
else:
raise Exception(f"Failed request, Status Code {response.status_code}")
## Extract Data
soup = BeautifulSoup(response.text, "html.parser")
div_cards = soup.select("div span")
for div_card in div_cards:
name = div_card.find("h3")
link = div_card.find("a")
if not name or not link:
continue
result_number += 1
ranking = result_number
search_data = SearchData(
name=name.text,
url=link.get("href"),
rank=ranking
)
data_pipeline.add_data(search_data)
logger.info(f"Successfully parsed data from: {url}")
success = True
except Exception as e:
logger.error(f"An error occurred while processing page {url}: {e}")
logger.info(f"Retrying request for page: {url}, retries left {retries-tries}")
if not success:
raise Exception(f"Max Retries exceeded: {retries}")
def start_scrape(keyword, pages, location, data_pipeline=None, max_threads=5, retries=3):
with concurrent.futures.ThreadPoolExecutor(max_workers=max_threads) as executor:
executor.map(
scrape_search_results,
[keyword] * pages,
[location] * pages,
range(pages),
[data_pipeline] * pages,
[retries] * pages
)
if __name__ == "__main__":
MAX_RETRIES = 3
MAX_THREADS = 5
PAGES = 1
LOCATION = "us"
logger.info(f"Crawl starting...")
## INPUT ---> List of keywords to scrape
keyword_list = ["learn rust"]
aggregate_files = []
## Job Processes
for keyword in keyword_list:
filename = keyword.replace(" ", "-")
crawl_pipeline = DataPipeline(csv_filename=f"{filename}.csv")
start_scrape(keyword, PAGES, LOCATION, data_pipeline=crawl_pipeline, max_threads=MAX_THREADS, retries=MAX_RETRIES)
crawl_pipeline.close_pipeline()
aggregate_files.append(f"{filename}.csv")
logger.info(f"Crawl complete.")Step 6: Production Run
Now, that we've built our crawler, let's run it in production. I'm going to set ourPAGES to 5 and time the operation. Here is our updated main.if __name__ == "__main__":
MAX_RETRIES = 3
MAX_THREADS = 5
PAGES = 5
LOCATION = "us"
logger.info(f"Crawl starting...")
## INPUT ---> List of keywords to scrape
keyword_list = ["learn rust"]
aggregate_files = []
## Job Processes
for keyword in keyword_list:
filename = keyword.replace(" ", "-")
crawl_pipeline = DataPipeline(csv_filename=f"{filename}.csv")
start_scrape(keyword, PAGES, LOCATION, data_pipeline=crawl_pipeline, max_threads=MAX_THREADS, retries=MAX_RETRIES)
crawl_pipeline.close_pipeline()
aggregate_files.append(f"{filename}.csv")
logger.info(f"Crawl complete.")main:MAX_RETRIESPAGESMAX_THREADSLOCATIONkeyword_list
Build A Quora Scraper
To scrape Quora posts, we need to build a scraper that performs the following steps:- Read the CSV from our crawler.
- Parse each post from the CSV.
- Store the data that we extracted in the parse.
- Perform steps 2 and 3 with concurrency so we can process multiple posts at the same time.
- Integrate with a proxy in order to get past anti-bots.
Step 1: Create Simple Business Data Parser
Like we did earlier, we'll start with a basic parsing function. This parser will extract reply information from each post.With Quora particularly, our data is incredibly nested and there are a ton of replies that show up that aren't specific to the question being asked.This parser is going to filter out the promoted and related replies so that we only get direct replies to the questions being asked.Here is the parsing function we'll start with.def process_post(row, location, retries=3):
url = row["url"]
tries = 0
success = False
while tries <= retries and not success:
response = requests.get(url, location=location)
try:
if response.status_code == 200:
logger.info(f"Status: {response.status_code}")
soup = BeautifulSoup(response.text, "html.parser")
main_content = soup.select_one("div[id='mainContent']")
answer_cards = main_content.select("div div div div div[class='q-box']")
last_seen_name = ""
for answer_card in answer_cards:
excluded_words = ["All related", "Recommended"]
array = answer_card.text.split("·")
if len(array) < 3:
continue
promoted = "Promoted" in array[0]
related = "Related" in array[2][0:30] or "Related" in array[-2][0:30]
repeat_name = array[0] in last_seen_name or array[0] == last_seen_name
if promoted or related or repeat_name:
last_seen_name = array[0]
continue
print("name:", array[0])
print("reply:", array[-2])
success = True
else:
logger.warning(f"Failed Response: {response.status_code}")
raise Exception(f"Failed Request, status code: {response.status_code}")
except Exception as e:
logger.error(f"Exception thrown: {e}")
logger.warning(f"Failed to process page: {row['url']}")
logger.warning(f"Retries left: {retries-tries}")
tries += 1
if not success:
raise Exception(f"Max Retries exceeded: {retries}")
else:
logger.info(f"Successfully parsed: {row['url']}")- Find the
main_contentbox withsoup.select_one("div[id='mainContent']"). - We then find the
divelements holding our replies withmain_content.select("div div div div div[class='q-box']"). - We then split the text of these cards into an array.
- Arrays that are not long enough get filtered out.
- We then check to see if the card is a
"Promoted"or"Related"reply. - As long as this the reply is not promoted or related, we pull the
"name"and"reply"out of the string array.
Step 2: Loading URLs To Scrape
To use our parsing function, we need to load our URLs. We can do this by reading the CSV file generated by our crawler. Similar to ourstart_scrape() function, we'll create a new one, process_results(). This function will read the CSV file and call process_post() on all the rows from the file.Here is process_results().def process_results(csv_file, location, max_threads=5, retries=3):
logger.info(f"processing {csv_file}")
with open(csv_file, newline="") as file:
reader = list(csv.DictReader(file))
for row in reader:
process_post(row, location, retries=retries)for loop and then we run process_post() on each row from the file.Here is our full code up to this point.import os
import csv
import requests
import json
import logging
from urllib.parse import urlencode
from bs4 import BeautifulSoup
import concurrent.futures
from dataclasses import dataclass, field, fields, asdict
API_KEY = ""
with open("config.json", "r") as config_file:
config = json.load(config_file)
API_KEY = config["api_key"]
def get_scrapeops_url(url, location="us"):
payload = {
"api_key": API_KEY,
"url": url,
"country": location,
"wait": 2000
}
proxy_url = "https://proxy.scrapeops.io/v1/?" + urlencode(payload)
return proxy_url
## Logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class SearchData:
name: str = ""
url: str = ""
rank: int = 0
def __post_init__(self):
self.check_string_fields()
def check_string_fields(self):
for field in fields(self):
# Check string fields
if isinstance(getattr(self, field.name), str):
# If empty set default text
if getattr(self, field.name) == "":
setattr(self, field.name, f"No {field.name}")
continue
# Strip any trailing spaces, etc.
value = getattr(self, field.name)
setattr(self, field.name, value.strip())
class DataPipeline:
def __init__(self, csv_filename="", storage_queue_limit=50):
self.names_seen = []
self.storage_queue = []
self.storage_queue_limit = storage_queue_limit
self.csv_filename = csv_filename
self.csv_file_open = False
def save_to_csv(self):
self.csv_file_open = True
data_to_save = []
data_to_save.extend(self.storage_queue)
self.storage_queue.clear()
if not data_to_save:
return
keys = [field.name for field in fields(data_to_save[0])]
file_exists = os.path.isfile(self.csv_filename) and os.path.getsize(self.csv_filename) > 0
with open(self.csv_filename, mode="a", newline="", encoding="utf-8") as output_file:
writer = csv.DictWriter(output_file, fieldnames=keys)
if not file_exists:
writer.writeheader()
for item in data_to_save:
writer.writerow(asdict(item))
self.csv_file_open = False
def is_duplicate(self, input_data):
if input_data.name in self.names_seen:
logger.warning(f"Duplicate item found: {input_data.name}. Item dropped.")
return True
self.names_seen.append(input_data.name)
return False
def add_data(self, scraped_data):
if self.is_duplicate(scraped_data) == False:
self.storage_queue.append(scraped_data)
if len(self.storage_queue) >= self.storage_queue_limit and self.csv_file_open == False:
self.save_to_csv()
def close_pipeline(self):
if self.csv_file_open:
time.sleep(3)
if len(self.storage_queue) > 0:
self.save_to_csv()
def scrape_search_results(keyword, location, page_number, data_pipeline=None, retries=3):
formatted_keyword = keyword.replace(" ", "+")
result_number = page_number * 10
url = f"https://www.google.com/search?q={formatted_keyword}%20site%3Aquora.com&start={result_number}"
tries = 0
success = False
while tries <= retries and not success:
try:
scrapeops_proxy_url = get_scrapeops_url(url, location=location)
response = requests.get(scrapeops_proxy_url)
logger.info(f"Recieved [{response.status_code}] from: {url}")
if response.status_code == 200:
success = True
else:
raise Exception(f"Failed request, Status Code {response.status_code}")
## Extract Data
soup = BeautifulSoup(response.text, "html.parser")
div_cards = soup.select("div span")
for div_card in div_cards:
name = div_card.find("h3")
link = div_card.find("a")
if not name or not link:
continue
result_number += 1
ranking = result_number
search_data = SearchData(
name=name.text,
url=link.get("href"),
rank=ranking
)
data_pipeline.add_data(search_data)
logger.info(f"Successfully parsed data from: {url}")
success = True
except Exception as e:
logger.error(f"An error occurred while processing page {url}: {e}")
logger.info(f"Retrying request for page: {url}, retries left {retries-tries}")
if not success:
raise Exception(f"Max Retries exceeded: {retries}")
def start_scrape(keyword, pages, location, data_pipeline=None, max_threads=5, retries=3):
with concurrent.futures.ThreadPoolExecutor(max_workers=max_threads) as executor:
executor.map(
scrape_search_results,
[keyword] * pages,
[location] * pages,
range(pages),
[data_pipeline] * pages,
[retries] * pages
)
def process_post(row, location, retries=3):
url = row["url"]
tries = 0
success = False
while tries <= retries and not success:
response = requests.get(url, location=location)
try:
if response.status_code == 200:
logger.info(f"Status: {response.status_code}")
soup = BeautifulSoup(response.text, "html.parser")
main_content = soup.select_one("div[id='mainContent']")
answer_cards = main_content.select("div div div div div[class='q-box']")
last_seen_name = ""
for answer_card in answer_cards:
excluded_words = ["All related", "Recommended"]
array = answer_card.text.split("·")
if len(array) < 3:
continue
promoted = "Promoted" in array[0]
related = "Related" in array[2][0:30] or "Related" in array[-2][0:30]
repeat_name = array[0] in last_seen_name or array[0] == last_seen_name
if promoted or related or repeat_name:
last_seen_name = array[0]
continue
print("name:", array[0])
print("reply:", array[-2])
success = True
else:
logger.warning(f"Failed Response: {response.status_code}")
raise Exception(f"Failed Request, status code: {response.status_code}")
except Exception as e:
logger.error(f"Exception thrown: {e}")
logger.warning(f"Failed to process page: {row['url']}")
logger.warning(f"Retries left: {retries-tries}")
tries += 1
if not success:
raise Exception(f"Max Retries exceeded: {retries}")
else:
logger.info(f"Successfully parsed: {row['url']}")
def process_results(csv_file, location, max_threads=5, retries=3):
logger.info(f"processing {csv_file}")
with open(csv_file, newline="") as file:
reader = list(csv.DictReader(file))
for row in reader:
process_post(row, location, retries=retries)
if __name__ == "__main__":
MAX_RETRIES = 3
MAX_THREADS = 5
PAGES = 1
LOCATION = "us"
logger.info(f"Crawl starting...")
## INPUT ---> List of keywords to scrape
keyword_list = ["learn rust"]
aggregate_files = []
## Job Processes
for keyword in keyword_list:
filename = keyword.replace(" ", "-")
crawl_pipeline = DataPipeline(csv_filename=f"{filename}.csv")
start_scrape(keyword, PAGES, LOCATION, data_pipeline=crawl_pipeline, max_threads=MAX_THREADS, retries=MAX_RETRIES)
crawl_pipeline.close_pipeline()
aggregate_files.append(f"{filename}.csv")
logger.info(f"Crawl complete.")
for file in aggregate_files:
process_results(file, LOCATION, max_threads=MAX_THREADS, retries=MAX_RETRIES)Step 3: Storing the Scraped Data
We're now reading a CSV file and properly parsing the information from it. Now, we need to store the data. Sound familiar?This is going to be relatively easier than last time. We already have ourDataPipeline, we just need another dataclass.In this section, we'll create a ReplyData class. It's going to hold a name and the content of the reply.Here is our new dataclass, ReplyData.@dataclass
class ReplyData:
name: str = ""
reply: str = ""
def __post_init__(self):
self.check_string_fields()
def check_string_fields(self):
for field in fields(self):
# Check string fields
if isinstance(getattr(self, field.name), str):
# If empty set default text
if getattr(self, field.name) == "":
setattr(self, field.name, f"No {field.name}")
continue
# Strip any trailing spaces, etc.
value = getattr(self, field.name)
setattr(self, field.name, value.strip())DataPipeline instance during our parsing function.import os
import csv
import requests
import json
import logging
from urllib.parse import urlencode
from bs4 import BeautifulSoup
import concurrent.futures
from dataclasses import dataclass, field, fields, asdict
API_KEY = ""
with open("config.json", "r") as config_file:
config = json.load(config_file)
API_KEY = config["api_key"]
def get_scrapeops_url(url, location="us"):
payload = {
"api_key": API_KEY,
"url": url,
"country": location,
"wait": 2000
}
proxy_url = "https://proxy.scrapeops.io/v1/?" + urlencode(payload)
return proxy_url
## Logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class SearchData:
name: str = ""
url: str = ""
rank: int = 0
def __post_init__(self):
self.check_string_fields()
def check_string_fields(self):
for field in fields(self):
# Check string fields
if isinstance(getattr(self, field.name), str):
# If empty set default text
if getattr(self, field.name) == "":
setattr(self, field.name, f"No {field.name}")
continue
# Strip any trailing spaces, etc.
value = getattr(self, field.name)
setattr(self, field.name, value.strip())
@dataclass
class ReplyData:
name: str = ""
reply: str = ""
def __post_init__(self):
self.check_string_fields()
def check_string_fields(self):
for field in fields(self):
# Check string fields
if isinstance(getattr(self, field.name), str):
# If empty set default text
if getattr(self, field.name) == "":
setattr(self, field.name, f"No {field.name}")
continue
# Strip any trailing spaces, etc.
value = getattr(self, field.name)
setattr(self, field.name, value.strip())
class DataPipeline:
def __init__(self, csv_filename="", storage_queue_limit=50):
self.names_seen = []
self.storage_queue = []
self.storage_queue_limit = storage_queue_limit
self.csv_filename = csv_filename
self.csv_file_open = False
def save_to_csv(self):
self.csv_file_open = True
data_to_save = []
data_to_save.extend(self.storage_queue)
self.storage_queue.clear()
if not data_to_save:
return
keys = [field.name for field in fields(data_to_save[0])]
file_exists = os.path.isfile(self.csv_filename) and os.path.getsize(self.csv_filename) > 0
with open(self.csv_filename, mode="a", newline="", encoding="utf-8") as output_file:
writer = csv.DictWriter(output_file, fieldnames=keys)
if not file_exists:
writer.writeheader()
for item in data_to_save:
writer.writerow(asdict(item))
self.csv_file_open = False
def is_duplicate(self, input_data):
if input_data.name in self.names_seen:
logger.warning(f"Duplicate item found: {input_data.name}. Item dropped.")
return True
self.names_seen.append(input_data.name)
return False
def add_data(self, scraped_data):
if self.is_duplicate(scraped_data) == False:
self.storage_queue.append(scraped_data)
if len(self.storage_queue) >= self.storage_queue_limit and self.csv_file_open == False:
self.save_to_csv()
def close_pipeline(self):
if self.csv_file_open:
time.sleep(3)
if len(self.storage_queue) > 0:
self.save_to_csv()
def scrape_search_results(keyword, location, page_number, data_pipeline=None, retries=3):
formatted_keyword = keyword.replace(" ", "+")
result_number = page_number * 10
url = f"https://www.google.com/search?q={formatted_keyword}%20site%3Aquora.com&start={result_number}"
tries = 0
success = False
while tries <= retries and not success:
try:
scrapeops_proxy_url = get_scrapeops_url(url, location=location)
response = requests.get(scrapeops_proxy_url)
logger.info(f"Recieved [{response.status_code}] from: {url}")
if response.status_code == 200:
success = True
else:
raise Exception(f"Failed request, Status Code {response.status_code}")
## Extract Data
soup = BeautifulSoup(response.text, "html.parser")
div_cards = soup.select("div span")
for div_card in div_cards:
name = div_card.find("h3")
link = div_card.find("a")
if not name or not link:
continue
result_number += 1
ranking = result_number
search_data = SearchData(
name=name.text,
url=link.get("href"),
rank=ranking
)
data_pipeline.add_data(search_data)
logger.info(f"Successfully parsed data from: {url}")
success = True
except Exception as e:
logger.error(f"An error occurred while processing page {url}: {e}")
logger.info(f"Retrying request for page: {url}, retries left {retries-tries}")
if not success:
raise Exception(f"Max Retries exceeded: {retries}")
def start_scrape(keyword, pages, location, data_pipeline=None, max_threads=5, retries=3):
with concurrent.futures.ThreadPoolExecutor(max_workers=max_threads) as executor:
executor.map(
scrape_search_results,
[keyword] * pages,
[location] * pages,
range(pages),
[data_pipeline] * pages,
[retries] * pages
)
def process_post(row, location, retries=3):
url = row["url"]
tries = 0
success = False
while tries <= retries and not success:
response = requests.get(url, location=location)
try:
if response.status_code == 200:
logger.info(f"Status: {response.status_code}")
soup = BeautifulSoup(response.text, "html.parser")
main_content = soup.select_one("div[id='mainContent']")
answer_cards = main_content.select("div div div div div[class='q-box']")
answer_pipeline = DataPipeline(csv_filename=f"{row['name'].replace(' ', '-')}.csv")
last_seen_name = ""
for answer_card in answer_cards:
excluded_words = ["All related", "Recommended"]
array = answer_card.text.split("·")
if len(array) < 3:
continue
promoted = "Promoted" in array[0]
related = "Related" in array[2][0:30] or "Related" in array[-2][0:30]
repeat_name = array[0] in last_seen_name or array[0] == last_seen_name
if promoted or related or repeat_name:
last_seen_name = array[0]
continue
reply_data = ReplyData(
name=array[0],
reply=array[-2]
)
answer_pipeline.add_data(reply_data)
answer_pipeline.close_pipeline()
success = True
else:
logger.warning(f"Failed Response: {response.status_code}")
raise Exception(f"Failed Request, status code: {response.status_code}")
except Exception as e:
logger.error(f"Exception thrown: {e}")
logger.warning(f"Failed to process page: {row['url']}")
logger.warning(f"Retries left: {retries-tries}")
tries += 1
if not success:
raise Exception(f"Max Retries exceeded: {retries}")
else:
logger.info(f"Successfully parsed: {row['url']}")
def process_results(csv_file, location, max_threads=5, retries=3):
logger.info(f"processing {csv_file}")
with open(csv_file, newline="") as file:
reader = list(csv.DictReader(file))
for row in reader:
process_post(row, location, retries=retries)
if __name__ == "__main__":
MAX_RETRIES = 3
MAX_THREADS = 5
PAGES = 1
LOCATION = "us"
logger.info(f"Crawl starting...")
## INPUT ---> List of keywords to scrape
keyword_list = ["learn rust"]
aggregate_files = []
## Job Processes
for keyword in keyword_list:
filename = keyword.replace(" ", "-")
crawl_pipeline = DataPipeline(csv_filename=f"{filename}.csv")
start_scrape(keyword, PAGES, LOCATION, data_pipeline=crawl_pipeline, max_threads=MAX_THREADS, retries=MAX_RETRIES)
crawl_pipeline.close_pipeline()
aggregate_files.append(f"{filename}.csv")
logger.info(f"Crawl complete.")
for file in aggregate_files:
process_results(file, LOCATION, max_threads=MAX_THREADS, retries=MAX_RETRIES)ReplyData and another DataPipeline, we can now pipe reply data straight to a CSV file.Step 4: Adding Concurrency
We're now at a point where we need to add concurrency. To accomplish this, we useThreadPoolExecutor just like we did earlier. We pass in process_post as our first argument, and then we pass subsequent arguments in as arrays, just like we did before.Here is our refactored process_results() function.def process_results(csv_file, location, max_threads=5, retries=3):
logger.info(f"processing {csv_file}")
with open(csv_file, newline="") as file:
reader = list(csv.DictReader(file))
with concurrent.futures.ThreadPoolExecutor(max_workers=max_threads) as executor:
executor.map(
process_post,
reader,
[location] * len(reader),
[retries] * len(reader)
)process_postis the function we want to run on each available thread.- All arguments to
process_postget passed intoexecutor.map()as arrays.
Step 5: Bypassing Anti-Bots
At this point, bypassing anti-bots is a cinch. We already have all the infrastructure we need. All we need to do is change a single line in our parsing function.response = requests.get(get_scrapeops_url(url, location=location))
import os
import csv
import requests
import json
import logging
from urllib.parse import urlencode
from bs4 import BeautifulSoup
import concurrent.futures
from dataclasses import dataclass, field, fields, asdict
API_KEY = ""
with open("config.json", "r") as config_file:
config = json.load(config_file)
API_KEY = config["api_key"]
def get_scrapeops_url(url, location="us"):
payload = {
"api_key": API_KEY,
"url": url,
"country": location,
"wait": 2000
}
proxy_url = "https://proxy.scrapeops.io/v1/?" + urlencode(payload)
return proxy_url
## Logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class SearchData:
name: str = ""
url: str = ""
rank: int = 0
def __post_init__(self):
self.check_string_fields()
def check_string_fields(self):
for field in fields(self):
# Check string fields
if isinstance(getattr(self, field.name), str):
# If empty set default text
if getattr(self, field.name) == "":
setattr(self, field.name, f"No {field.name}")
continue
# Strip any trailing spaces, etc.
value = getattr(self, field.name)
setattr(self, field.name, value.strip())
@dataclass
class ReplyData:
name: str = ""
reply: str = ""
def __post_init__(self):
self.check_string_fields()
def check_string_fields(self):
for field in fields(self):
# Check string fields
if isinstance(getattr(self, field.name), str):
# If empty set default text
if getattr(self, field.name) == "":
setattr(self, field.name, f"No {field.name}")
continue
# Strip any trailing spaces, etc.
value = getattr(self, field.name)
setattr(self, field.name, value.strip())
class DataPipeline:
def __init__(self, csv_filename="", storage_queue_limit=50):
self.names_seen = []
self.storage_queue = []
self.storage_queue_limit = storage_queue_limit
self.csv_filename = csv_filename
self.csv_file_open = False
def save_to_csv(self):
self.csv_file_open = True
data_to_save = []
data_to_save.extend(self.storage_queue)
self.storage_queue.clear()
if not data_to_save:
return
keys = [field.name for field in fields(data_to_save[0])]
file_exists = os.path.isfile(self.csv_filename) and os.path.getsize(self.csv_filename) > 0
with open(self.csv_filename, mode="a", newline="", encoding="utf-8") as output_file:
writer = csv.DictWriter(output_file, fieldnames=keys)
if not file_exists:
writer.writeheader()
for item in data_to_save:
writer.writerow(asdict(item))
self.csv_file_open = False
def is_duplicate(self, input_data):
if input_data.name in self.names_seen:
logger.warning(f"Duplicate item found: {input_data.name}. Item dropped.")
return True
self.names_seen.append(input_data.name)
return False
def add_data(self, scraped_data):
if self.is_duplicate(scraped_data) == False:
self.storage_queue.append(scraped_data)
if len(self.storage_queue) >= self.storage_queue_limit and self.csv_file_open == False:
self.save_to_csv()
def close_pipeline(self):
if self.csv_file_open:
time.sleep(3)
if len(self.storage_queue) > 0:
self.save_to_csv()
def scrape_search_results(keyword, location, page_number, data_pipeline=None, retries=3):
formatted_keyword = keyword.replace(" ", "+")
result_number = page_number * 10
url = f"https://www.google.com/search?q={formatted_keyword}%20site%3Aquora.com&start={result_number}"
tries = 0
success = False
while tries <= retries and not success:
try:
scrapeops_proxy_url = get_scrapeops_url(url, location=location)
response = requests.get(scrapeops_proxy_url)
logger.info(f"Recieved [{response.status_code}] from: {url}")
if response.status_code == 200:
success = True
else:
raise Exception(f"Failed request, Status Code {response.status_code}")
## Extract Data
soup = BeautifulSoup(response.text, "html.parser")
div_cards = soup.select("div span")
for div_card in div_cards:
name = div_card.find("h3")
link = div_card.find("a")
if not name or not link:
continue
result_number += 1
ranking = result_number
search_data = SearchData(
name=name.text,
url=link.get("href"),
rank=ranking
)
data_pipeline.add_data(search_data)
logger.info(f"Successfully parsed data from: {url}")
success = True
except Exception as e:
logger.error(f"An error occurred while processing page {url}: {e}")
logger.info(f"Retrying request for page: {url}, retries left {retries-tries}")
if not success:
raise Exception(f"Max Retries exceeded: {retries}")
def start_scrape(keyword, pages, location, data_pipeline=None, max_threads=5, retries=3):
with concurrent.futures.ThreadPoolExecutor(max_workers=max_threads) as executor:
executor.map(
scrape_search_results,
[keyword] * pages,
[location] * pages,
range(pages),
[data_pipeline] * pages,
[retries] * pages
)
def process_post(row, location, retries=3):
url = row["url"]
tries = 0
success = False
while tries <= retries and not success:
response = requests.get(get_scrapeops_url(url, location=location))
try:
if response.status_code == 200:
logger.info(f"Status: {response.status_code}")
soup = BeautifulSoup(response.text, "html.parser")
main_content = soup.select_one("div[id='mainContent']")
answer_cards = main_content.select("div div div div div[class='q-box']")
answer_pipeline = DataPipeline(csv_filename=f"{row['name'].replace(' ', '-')}.csv")
last_seen_name = ""
for answer_card in answer_cards:
excluded_words = ["All related", "Recommended"]
array = answer_card.text.split("·")
if len(array) < 3:
continue
promoted = "Promoted" in array[0]
related = "Related" in array[2][0:30] or "Related" in array[-2][0:30]
repeat_name = array[0] in last_seen_name or array[0] == last_seen_name
if promoted or related or repeat_name:
last_seen_name = array[0]
continue
reply_data = ReplyData(
name=array[0],
reply=array[-2]
)
answer_pipeline.add_data(reply_data)
answer_pipeline.close_pipeline()
success = True
else:
logger.warning(f"Failed Response: {response.status_code}")
raise Exception(f"Failed Request, status code: {response.status_code}")
except Exception as e:
logger.error(f"Exception thrown: {e}")
logger.warning(f"Failed to process page: {row['url']}")
logger.warning(f"Retries left: {retries-tries}")
tries += 1
if not success:
raise Exception(f"Max Retries exceeded: {retries}")
else:
logger.info(f"Successfully parsed: {row['url']}")
def process_results(csv_file, location, max_threads=5, retries=3):
logger.info(f"processing {csv_file}")
with open(csv_file, newline="") as file:
reader = list(csv.DictReader(file))
with concurrent.futures.ThreadPoolExecutor(max_workers=max_threads) as executor:
executor.map(
process_post,
reader,
[location] * len(reader),
[retries] * len(reader)
)
if __name__ == "__main__":
MAX_RETRIES = 3
MAX_THREADS = 5
PAGES = 1
LOCATION = "us"
logger.info(f"Crawl starting...")
## INPUT ---> List of keywords to scrape
keyword_list = ["learn rust"]
aggregate_files = []
## Job Processes
for keyword in keyword_list:
filename = keyword.replace(" ", "-")
crawl_pipeline = DataPipeline(csv_filename=f"{filename}.csv")
start_scrape(keyword, PAGES, LOCATION, data_pipeline=crawl_pipeline, max_threads=MAX_THREADS, retries=MAX_RETRIES)
crawl_pipeline.close_pipeline()
aggregate_files.append(f"{filename}.csv")
logger.info(f"Crawl complete.")
for file in aggregate_files:
process_results(file, LOCATION, max_threads=MAX_THREADS, retries=MAX_RETRIES)Step 6: Production Run
To test our full project in production, we'll use the same settings as before. If you need a refresher, here is ourmain.if __name__ == "__main__":
MAX_RETRIES = 3
MAX_THREADS = 5
PAGES = 5
LOCATION = "us"
logger.info(f"Crawl starting...")
## INPUT ---> List of keywords to scrape
keyword_list = ["learn rust"]
aggregate_files = []
## Job Processes
for keyword in keyword_list:
filename = keyword.replace(" ", "-")
crawl_pipeline = DataPipeline(csv_filename=f"{filename}.csv")
start_scrape(keyword, PAGES, LOCATION, data_pipeline=crawl_pipeline, max_threads=MAX_THREADS, retries=MAX_RETRIES)
crawl_pipeline.close_pipeline()
aggregate_files.append(f"{filename}.csv")
logger.info(f"Crawl complete.")
for file in aggregate_files:
process_results(file, LOCATION, max_threads=MAX_THREADS, retries=MAX_RETRIES)Legal and Ethical Considerations
When scraping the web, you need to pay attention to your target site's Terms of Service and theirrobots.txt. Legal or not, when you violate a site's terms, you can get suspended or even permanently banned.Quora's terms are available to view here. Here is their robots.txt.Public data is generally free to scrape. Public data is any data that is not gated behind a login page or some other type of authentication.When scraping private data, you are subject to a site's terms and privacy laws in the site's jurisdiction. If you don't know if your scraper is legal, you should consult an attorney.Conclusion
Thank you for reading! You now have a solid grasp of Requests and BeautifulSoup. You're probably able to use CSS selectors effectively at this point as well. You've built a full scraping project that consists of parsing, pagination, data storage, concurrency, and proxy integration.If you'd like to learn more about the tech stack used in this article, check out these links below.How to Scrape Quora With Selenium
In this tutorial, we will learn how to build a Quora scraper using Selenium.- TLDR - How to Scrape Quora
- How To Architect Our Quora Scraper
- Understanding How To Scrape Quora
- Setting Up Our Quora Scraper Project
- Build A Quora Search Crawler
- Build A Quora Scraper
- Legal and Ethical Considerations
- Conclusion
- More Web Scraping Guides
Need help scraping the web?
Then check out ScrapeOps, the complete toolkit for web scraping.
TLDR - How to Scrape Quora with Selenium
Need to scrape Quora but don't have time to code? Use the scraper below!To quickly scrape Quora without coding from scratch, follow the steps below:- Write or paste the code provided to initiate the scraper.
-
You can configure various parameters:
MAX_THREADS: Controls the number of concurrent threads.MAX_RETRIES: Defines the retry attempts for failed requests.PAGES: Number of Google search result pages to scrape.LOCATION: Set the geographical region for search results.keyword_list: List of search terms to scrape.
-
Install the
virtualenvand install the necessary libraries. - After setting up, run the scraper, and data will be saved to CSV files.
import os
import csv
import json
import logging
import time
import string
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.keys import Keys
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, fields, asdict
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Selenium configuration
# Set the path to your ChromeDriver
CHROMEDRIVER_PATH = 'chromedriver.exe' # Adjust this to the actual path if necessary
# Configure the service to use the specified driver
service = Service(CHROMEDRIVER_PATH)
# Setup Chrome options for headless browsing
options = Options()
options.add_argument("--headless")
options.add_argument("--disable-gpu") # Required for headless mode in some environments
options.add_argument("--no-sandbox") # Especially useful for Linux environments
options.add_argument("--disable-dev-shm-usage") # Helps with resource issues on some systems
options.headless = True # Runs Chrome in headless mode (without GUI)
@dataclass
class SearchData:
name: str = ""
url: str = ""
rank: int = 0
def __post_init__(self):
self.check_string_fields()
def check_string_fields(self):
for field in fields(self):
value = getattr(self, field.name)
if isinstance(value, str):
if not value:
setattr(self, field.name, f"No {field.name}")
else:
setattr(self, field.name, value.strip())
@dataclass
class ReplyData:
name: str = ""
reply: str = ""
def __post_init__(self):
self.check_string_fields()
def check_string_fields(self):
for field in fields(self):
value = getattr(self, field.name)
if isinstance(value, str):
if not value:
setattr(self, field.name, f"No {field.name}")
else:
setattr(self, field.name, value.strip())
class DataPipeline:
def __init__(self, csv_filename="", storage_queue_limit=50):
self.names_seen = []
self.storage_queue = []
self.storage_queue_limit = storage_queue_limit
self.csv_filename = csv_filename
self.csv_file_open = False
def save_to_csv(self):
try:
self.csv_file_open = True
data_to_save = self.storage_queue.copy()
self.storage_queue.clear()
if not data_to_save:
return
keys = [field.name for field in fields(data_to_save[0])]
logger.info("saving file data")
logger.info(keys)
# Filter out invalid characters from the filename
valid_chars = "-_.() %s%s" % (string.ascii_letters, string.digits)
valid_filename = ''.join(c for c in self.csv_filename if c in valid_chars)
logger.info(valid_filename)
file_exists = (
os.path.isfile(valid_filename) and os.path.getsize(valid_filename) > 0
)
if not file_exists:
with open(valid_filename, 'w', newline='') as output_file:
writer = csv.DictWriter(output_file, fieldnames=keys)
writer.writeheader()
with open(
valid_filename, mode="a", newline="", encoding="utf-8"
) as output_file:
writer = csv.DictWriter(output_file, fieldnames=keys)
for item in data_to_save:
writer.writerow(asdict(item))
self.csv_file_open = False
except Exception as e:
logger.error(f"Error saving csv {e}")
def is_duplicate(self, input_data):
if input_data.name in self.names_seen:
logger.warning(f"Duplicate item found: {input_data.name}. Item dropped.")
return True
self.names_seen.append(input_data.name)
return False
def add_data(self, scraped_data):
logger.info("adding data")
logger.info(scraped_data)
if not self.is_duplicate(scraped_data):
self.storage_queue.append(scraped_data)
if (
len(self.storage_queue) >= self.storage_queue_limit
and not self.csv_file_open
):
self.save_to_csv()
def close_pipeline(self):
if self.csv_file_open:
time.sleep(3)
if self.storage_queue:
self.save_to_csv()
def scrape_search_results(keyword, page_number, data_pipeline=None, retries=3):
# Use a context manager to ensure the driver is properly closed
with webdriver.Chrome(service=service, options=options) as driver:
formatted_keyword = keyword.replace(" ", "+")
result_number = page_number * 10
logger.info(f"page {page_number}")
url = f"https://www.google.com/search?q={formatted_keyword}%20site%3Aquora.com&start={result_number}"
success = False
tries = 0
while tries < retries and not success:
try:
driver.get(url)
logger.info(f"Accessing {url}")
# Use explicit wait to ensure elements are loaded
wait = WebDriverWait(driver, 10)
wait.until(EC.presence_of_element_located((By.ID, "rso")))
# Extract search result cards
for i in range(1, 11):
try:
# Attempt primary XPath
name = driver.find_element(By.XPATH, f"//*[@id='rso']/div[{i}]/div/div/div[1]/div/div/span/a/h3").text
link = driver.find_element(By.XPATH, f"//*[@id='rso']/div[{i}]/div/div/div[1]/div/div/span/a").get_attribute("href")
except:
try:
# Fallback XPath
name = driver.find_element(By.XPATH, f"//*[@id='rso']/div[{i}]/div/div/div/div[1]/div/div/span/a/h3").text
link = driver.find_element(By.XPATH, f"//*[@id='rso']/div[{i}]/div/div/div/div[1]/div/div/span/a").get_attribute("href")
except Exception as e:
continue
search_data = SearchData(
name=name,
url=link,
rank=result_number + i # Increment rank per result
)
data_pipeline.add_data(search_data)
logger.info(f"Successfully parsed data from: {url}")
success = True
except Exception as e:
logger.error(f"An error occurred while processing {url}: {e}")
tries += 1
if tries >= retries:
logger.error(f"Max retries exceeded for {url}")
else:
logger.info(f"Retrying {url} ({tries}/{retries})")
time.sleep(2)
logger.info(f"Storage queue length after page {page_number}: {len(data_pipeline.storage_queue)}")
def start_scrape(
keyword, pages, data_pipeline=None, max_threads=5, retries=3
):
with ThreadPoolExecutor(max_workers=max_threads) as executor:
futures = []
for page in range(pages):
# No need to pass the driver anymore, each thread will create its own
futures.append(
executor.submit(
scrape_search_results,
keyword,
page,
data_pipeline,
retries,
)
)
# Ensure all threads complete
for future in futures:
future.result() # This blocks until the thread finishes
def process_post(row, retries=3):
with webdriver.Chrome(service=service, options=options) as driver:
logger.info(f"Processing row: {row}")
url = row.get("url")
if not url:
logger.error(f"No URL found in row: {row}")
return
logger.info(f"Processing URL: {url}")
success = False
tries = 0
while tries < retries and not success:
try:
driver.get(url)
logger.info(f"Accessing {url}")
# Use explicit wait to ensure main content is loaded
wait = WebDriverWait(driver, 10)
main_content = wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, "div[id='mainContent']")))
# Extract answer cards
answer_cards = main_content.find_elements(By.CSS_SELECTOR, "div.q-click-wrapper")
if not answer_cards:
logger.warning(f"No answer cards found at {url}")
# Initialize a new DataPipeline for replies
if 'name' not in row:
logger.error(f"'name' key missing in row: {row}")
break
answer_pipeline = DataPipeline(
csv_filename=f"{row['name'].replace(' ', '-')}.csv"
)
last_seen_name = ""
for answer_card in answer_cards:
try:
name_element = answer_card.find_element(By.CSS_SELECTOR, "div.q-relative")
name = name_element.text.replace("\n", "").strip()
reply_element = answer_card.find_element(By.CSS_SELECTOR, "div.spacing_log_answer_content")
reply = reply_element.text.strip()
if "Sponsored" in name:
continue
if "Related questions" in name:
break
if name == last_seen_name:
continue
last_seen_name = name
reply_data = ReplyData(name=name, reply=reply)
answer_pipeline.add_data(reply_data)
except Exception as e:
continue
answer_pipeline.close_pipeline()
success = True
except Exception as e:
logger.error(f"Exception thrown while processing {url}: {e}")
tries += 1
if tries >= retries:
logger.error(f"Max retries exceeded for {url}")
else:
logger.info(f"Retrying {url} ({tries}/{retries})")
time.sleep(2)
def process_results(csv_file, max_threads=5, retries=3):
with open(csv_file, newline="") as file:
reader = list(csv.DictReader(file))
logger.info(f"file opened")
with ThreadPoolExecutor(max_workers=max_threads) as executor:
for row in reader:
executor.submit(process_post, row, retries)
if __name__ == "__main__":
MAX_RETRIES = 3
MAX_THREADS = 5
PAGES = 5
logger.info(f"Crawl starting...")
# INPUT ---> List of keywords to scrape
keyword_list = ["learn rust"]
aggregate_files = []
# Job Processes
for keyword in keyword_list:
filename = keyword.replace(" ", "-")
crawl_pipeline = DataPipeline(csv_filename=f"{filename}.csv")
start_scrape(keyword, PAGES, data_pipeline=crawl_pipeline, max_threads=MAX_THREADS, retries=MAX_RETRIES)
crawl_pipeline.close_pipeline()
aggregate_files.append(f"{filename}.csv")
logger.info(f"Crawl complete.")
for file in aggregate_files:
process_results(file, max_threads=MAX_THREADS, retries=MAX_RETRIES)How To Architect Our Quora Scraper
Our project consists of two main components: a crawler and a scraper.- Crawler: It searches Quora through Google, extracts posts, and saves relevant data.
- Scraper: It reads the saved data, visits individual Quora posts, and scrapes detailed content.
- Perform a search on Quora through Google.
- Parse and extract search results, including pagination.
- Save data (post titles and links) efficiently.
- Execute concurrent searches on multiple result pages.
- Read the saved data from the CSV.
- Visit each individual Quora post and extract relevant information.
- Store extracted data in a structured format.
- Use concurrency to speed up scraping multiple posts.
Understanding How To Scrape Quora
Scraping Quora is unique compared to scraping other websites due to its frequent usage of dynamic content, anti-bot mechanisms, and requiring users to be logged in for most interactions.However, by leveraging Google search to find Quora posts and extracting the content using Selenium, we can bypass the need for an account and scrape publicly available Quora posts indirectly.Here’s a breakdown of how we scrape Quora:Step 1: How To Request Quora Pages
Directly scraping Quora pages can be difficult because accessing them usually requires logging in. To circumvent this, we scrape Quora indirectly by querying Google search results for Quora pages.If we just search Quora, on google it gives us an option to search it through there without going to the site.Here we query Google with the following structure to find relevant Quora pages:https://www.google.com/search?q={formatted_keyword}%20site%3Aquora.com
{formatted_keyword} is the term you're searching for on Quora.For example, searching for "learn Rust" on Quora via Google would look like this:https://www.google.com/search?q=learn+rust+site%3Aquora.com
quora.com.Step 2: How To Extract Data From Quora Results and Pages
Once you have retrieved the Google search results, you will need to extract URLs from the results and scrape the actual Quora pages for detailed answers.Finding the Correct XPaths or CSS Selectors
You can use Chrome DevTools (right-click on a webpage, then click “Inspect”) to find the correct XPaths or CSS selectors for the elements you want to extract.1. Find XPath for Quora Search Results:
- For each search result in Google, you’ll need to locate the post titles and URLs.
- To locate the post title in a Google search result, right-click on the title element in Chrome DevTools and select “CopXPath”.
- This might give you an XPath like:
//*[@id='rso']/div[1]/div/div[1]/a/h3
- Use similar techniques to extract the URL.
Screenshot 42. Find the Main Content and Replies on Quora:
Once you have the Quora post URLs, you’ll need to scrape the actual content and replies on each Quora page. In Quora posts, the answers are often deeply nested withindiv tags.Use the following steps to locate elements:- Inspect the page, and right-click the area containing the answers.
- Copy the CSS selector or XPath of the answer's container. For example, Quora uses a class like
q-click-wrapperfor answers.
Screenshot 5Step 3: How To Control Pagination:
When scraping multiple Google search result pages, you’ll need to control pagination by updating the search URL’sstart parameter.For example:- Page 1: Returns the first 10 results.
https://www.google.com/search?q=learn+rust+site%3Aquora.com&start=0
- Page 2: Returns the next 10 results.
https://www.google.com/search?q=learn+rust+site%3Aquora.com&start=10
start value by 10, you can paginate through all the results.Setting Up Our Quora Scraper Project
To get started with scraping Quora using Selenium and ScrapeOps, follow the steps below to set up the project environment, install dependencies, and configure your WebDriver.Create a New Project Foldermkdir quora-scraper cd quora-scraper
python -m venv venv source venv/bin/activate # On Windows use: venv\Scripts\activate
Selenium for browser automation, WebDriverWait for adding wait until an element appears on the browser and any other necessary libraries. Install them using pip:pip install selenium pip install WebDriverWait
-
Download ChromeDriver:
- Go to the ChromeDriver download page.
- Make sure to download the version that matches your installed version of Google Chrome.
-
Move ChromeDriver to Project Path:
- Once downloaded, place
chromedriver.exein your project folder or somewhere accessible in your system’sPATH.
- Once downloaded, place
CHROMEDRIVER_PATH and set up the Service:CHROMEDRIVER_PATH = 'chromedriver.exe' # Adjust this to the actual path if chromedriver file is not in the current directory
Build A Quora Search Crawler
Step 1: Create Simple Search Data Parser
Create a parser that extracts Quora post titles and links from Google search results. UseSelenium to select the HTML elements (h3 for titles and a for links).import os
import csv
import json
import logging
import time
import string
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.keys import Keys
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, fields, asdict
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Selenium configuration
# Set the path to your ChromeDriver
CHROMEDRIVER_PATH = 'chromedriver.exe' # Adjust this to the actual path if necessary
# Configure the service to use the specified driver
service = Service(CHROMEDRIVER_PATH)
# Setup Chrome options for headless browsing
options = Options()
options.add_argument("--headless")
options.add_argument("--disable-gpu") # Required for headless mode in some environments
options.add_argument("--no-sandbox") # Especially useful for Linux environments
options.add_argument("--disable-dev-shm-usage") # Helps with resource issues on some systems
options.headless = True # Runs Chrome in headless mode (without GUI)
def scrape_search_results(keyword, data_pipeline=None, retries=3):
# Use a context manager to ensure the driver is properly closed
with webdriver.Chrome(service=service, options=options) as driver:
formatted_keyword = keyword.replace(" ", "+")
result_number = 0
logger.info(f"page {page_number}")
url = f"https://www.google.com/search?q={formatted_keyword}%20site%3Aquora.com&start={result_number}"
success = False
tries = 0
while tries < retries and not success:
try:
driver.get(url)
logger.info(f"Accessing {url}")
# Use explicit wait to ensure elements are loaded
wait = WebDriverWait(driver, 10)
wait.until(EC.presence_of_element_located((By.ID, "rso")))
# Extract search result cards
for i in range(1, 11):
try:
# Attempt primary XPath
name = driver.find_element(By.XPATH, f"//*[@id='rso']/div[{i}]/div/div/div[1]/div/div/span/a/h3").text
link = driver.find_element(By.XPATH, f"//*[@id='rso']/div[{i}]/div/div/div[1]/div/div/span/a").get_attribute("href")
except:
try:
# Fallback XPath
name = driver.find_element(By.XPATH, f"//*[@id='rso']/div[{i}]/div/div/div/div[1]/div/div/span/a/h3").text
link = driver.find_element(By.XPATH, f"//*[@id='rso']/div[{i}]/div/div/div/div[1]/div/div/span/a").get_attribute("href")
except Exception as e:
continue
search_data = SearchData(
name=name,
url=link,
rank=result_number + i # Increment rank per result
)
data_pipeline.add_data(search_data)
logger.info(f"Successfully parsed data from: {url}")
success = True
except Exception as e:
logger.error(f"An error occurred while processing {url}: {e}")
tries += 1
if tries >= retries:
logger.error(f"Max retries exceeded for {url}")
else:
logger.info(f"Retrying {url} ({tries}/{retries})")
time.sleep(2)
logger.info(f"Storage queue length after page {page_number}: {len(data_pipeline.storage_queue)}")
if __name__ == "__main__":
MAX_RETRIES = 3
MAX_THREADS = 5
PAGES = 5
logger.info(f"Crawl starting...")
# INPUT ---> List of keywords to scrape
keyword_list = ["learn rust"]
aggregate_files = []
# Job Processes
for keyword in keyword_list:
filename = keyword.replace(" ", "-")
scrape_search_results(keyword, retries=MAX_RETRIES)Step 2: Add Pagination
Modify the search URL to paginate through results by adjusting thestart parameter:result_number = page_number * 10
url = f"https://www.google.com/search?q={formatted_keyword}%20site%3Aquora.com&start={result_number}"import os
import csv
import json
import logging
import time
import string
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.keys import Keys
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, fields, asdict
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Selenium configuration
# Set the path to your ChromeDriver
CHROMEDRIVER_PATH = 'chromedriver.exe' # Adjust this to the actual path if necessary
# Configure the service to use the specified driver
service = Service(CHROMEDRIVER_PATH)
# Setup Chrome options for headless browsing
options = Options()
options.add_argument("--headless")
options.add_argument("--disable-gpu") # Required for headless mode in some environments
options.add_argument("--no-sandbox") # Especially useful for Linux environments
options.add_argument("--disable-dev-shm-usage") # Helps with resource issues on some systems
options.headless = True # Runs Chrome in headless mode (without GUI)
def scrape_search_results(keyword, page_number, data_pipeline=None, retries=3):
# Use a context manager to ensure the driver is properly closed
with webdriver.Chrome(service=service, options=options) as driver:
formatted_keyword = keyword.replace(" ", "+")
result_number = page_number * 10
logger.info(f"page {page_number}")
url = f"https://www.google.com/search?q={formatted_keyword}%20site%3Aquora.com&start={result_number}"
success = False
tries = 0
while tries < retries and not success:
try:
driver.get(url)
logger.info(f"Accessing {url}")
# Use explicit wait to ensure elements are loaded
wait = WebDriverWait(driver, 10)
wait.until(EC.presence_of_element_located((By.ID, "rso")))
# Extract search result cards
for i in range(1, 11):
try:
# Attempt primary XPath
name = driver.find_element(By.XPATH, f"//*[@id='rso']/div[{i}]/div/div/div[1]/div/div/span/a/h3").text
link = driver.find_element(By.XPATH, f"//*[@id='rso']/div[{i}]/div/div/div[1]/div/div/span/a").get_attribute("href")
except:
try:
# Fallback XPath
name = driver.find_element(By.XPATH, f"//*[@id='rso']/div[{i}]/div/div/div/div[1]/div/div/span/a/h3").text
link = driver.find_element(By.XPATH, f"//*[@id='rso']/div[{i}]/div/div/div/div[1]/div/div/span/a").get_attribute("href")
except Exception as e:
continue
search_data = SearchData(
name=name,
url=link,
rank=result_number + i # Increment rank per result
)
data_pipeline.add_data(search_data)
logger.info(f"Successfully parsed data from: {url}")
success = True
except Exception as e:
logger.error(f"An error occurred while processing {url}: {e}")
tries += 1
if tries >= retries:
logger.error(f"Max retries exceeded for {url}")
else:
logger.info(f"Retrying {url} ({tries}/{retries})")
time.sleep(2)
logger.info(f"Storage queue length after page {page_number}: {len(data_pipeline.storage_queue)}")
def start_scrape(keyword, pages, location, data_pipeline=None, retries=3):
for page in range(pages):
scrape_search_results(keyword, page, data_pipeline=data_pipeline, retries=retries)
if __name__ == "__main__":
MAX_RETRIES = 3
MAX_THREADS = 5
PAGES = 5
logger.info(f"Crawl starting...")
# INPUT ---> List of keywords to scrape
keyword_list = ["learn rust"]
aggregate_files = []
# Job Processes
for keyword in keyword_list:
filename = keyword.replace(" ", "-")
start_scrape(keyword, pages=PAGES, retries=MAX_RETRIES)Step 3: Storing the Scraped Data
Once the data is extracted from Quora or the Google search results, it is essential to store it efficiently and avoid any duplicates. The storing process involves two classes:SearchData and DataPipeline.These two classes work together to manage the data, ensure no duplicates are stored, and handle writing the data to CSV files.Let’s dive into how these classes work and how they facilitate the storage of scraped data.SearchData ClassThe SearchData class represents a single scraped search result from Quora. Each instance of this class stores the name (title), URL, and rank of a Quora search result. Using this class ensures that the scraped data is structured and can be processed systematically.Here is the structure of the SearchData class:@dataclass
class SearchData:
name: str = ""
url: str = ""
rank: int = 0
def __post_init__(self):
self.check_string_fields()
def check_string_fields(self):
for field in fields(self):
value = getattr(self, field.name)
if isinstance(value, str):
# If the field is a string and is empty, give it a default value
if not value:
setattr(self, field.name, f"No {field.name}")
else:
# Strip leading/trailing whitespace
setattr(self, field.name, value.strip())- Data Validation (
check_string_fields): After initializing the object, the__post_init__method checks the string fields (nameandurl) to ensure they are not empty. If a field is empty, it assigns a default value (No {field.name}), making sure that empty data doesn’t enter the pipeline. - Rank: The rank of each search result is tracked. This is useful for sorting or prioritizing data during analysis.
SearchData:search_data = SearchData(
name=name,
url=link,
rank=result_number + i # Increment rank per result
)DataPipeline ClassThe DataPipeline class is responsible for managing the collected data and writing it to a CSV file. It performs the following tasks:- Managing a storage queue: Holds the scraped data temporarily before writing it to the CSV file.
- Checking for duplicates: Prevents duplicate entries based on the
namefield. - Saving to a CSV file: Writes the data to a CSV file once the storage queue reaches the defined limit or when the process ends.
DataPipeline class:class DataPipeline:
def __init__(self, csv_filename="", storage_queue_limit=50):
self.names_seen = [] # Track names to avoid duplicates
self.storage_queue = [] # Temporary storage for scraped data
self.storage_queue_limit = storage_queue_limit # Limit before writing to CSV
self.csv_filename = csv_filename # Name of the CSV file
self.csv_file_open = False # Check if the file is open
def save_to_csv(self):
try:
self.csv_file_open = True
data_to_save = self.storage_queue.copy()
self.storage_queue.clear() # Clear the queue after copying
if not data_to_save:
return
keys = [field.name for field in fields(data_to_save[0])]
# Ensure the CSV filename is valid
valid_chars = "-_.() %s%s" % (string.ascii_letters, string.digits)
valid_filename = ''.join(c for c in self.csv_filename if c in valid_chars)
logger.info(valid_filename)
file_exists = os.path.isfile(valid_filename) and os.path.getsize(valid_filename) > 0
# Write the header if the file does not exist
if not file_exists:
with open(valid_filename, 'w', newline='') as output_file:
writer = csv.DictWriter(output_file, fieldnames=keys)
writer.writeheader()
# Append the data to the CSV
with open(valid_filename, mode="a", newline="", encoding="utf-8") as output_file:
writer = csv.DictWriter(output_file, fieldnames=keys)
for item in data_to_save:
writer.writerow(asdict(item))
self.csv_file_open = False
except Exception as e:
logger.error(f"Error saving CSV: {e}")
def is_duplicate(self, input_data):
if input_data.name in self.names_seen:
logger.warning(f"Duplicate item found: {input_data.name}. Item dropped.")
return True
self.names_seen.append(input_data.name)
return False
def add_data(self, scraped_data):
if not self.is_duplicate(scraped_data):
self.storage_queue.append(scraped_data)
if len(self.storage_queue) >= self.storage_queue_limit and not self.csv_file_open:
self.save_to_csv()
def close_pipeline(self):
if self.csv_file_open:
time.sleep(3)
if self.storage_queue:
self.save_to_csv()-
Queue-based Storage (
storage_queue): Scraped data is first stored in a temporary queue. Once the queue reaches the definedstorage_queue_limit(e.g., 50 entries), the data is saved to a CSV file. This avoids frequent I/O operations and optimizes performance. -
Duplicate Handling (
is_duplicate): Before adding new data to the queue, theis_duplicatemethod checks whether the data already exists by comparing thenamefield. If a duplicate is found, it logs a warning and skips the entry. -
CSV File Writing (
save_to_csv): When the queue is full or when the scraping process is complete, thesave_to_csvmethod is called to write the collected data to a CSV file. It also ensures that the filename is valid and does not contain any illegal characters. -
Closing the Pipeline (
close_pipeline): When scraping is finished, theclose_pipelinemethod ensures that any remaining data in the queue is written to the CSV file.
DataPipeline in the main,import os
import csv
import json
import logging
import time
import string
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.keys import Keys
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, fields, asdict
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Selenium configuration
# Set the path to your ChromeDriver
CHROMEDRIVER_PATH = 'chromedriver.exe' # Adjust this to the actual path if necessary
# Configure the service to use the specified driver
service = Service(CHROMEDRIVER_PATH)
# Setup Chrome options for headless browsing
options = Options()
options.add_argument("--headless")
options.add_argument("--disable-gpu") # Required for headless mode in some environments
options.add_argument("--no-sandbox") # Especially useful for Linux environments
options.add_argument("--disable-dev-shm-usage") # Helps with resource issues on some systems
options.headless = True # Runs Chrome in headless mode (without GUI)
@dataclass
class SearchData:
name: str = ""
url: str = ""
rank: int = 0
def __post_init__(self):
self.check_string_fields()
def check_string_fields(self):
for field in fields(self):
value = getattr(self, field.name)
if isinstance(value, str):
if not value:
setattr(self, field.name, f"No {field.name}")
else:
setattr(self, field.name, value.strip())
class DataPipeline:
def __init__(self, csv_filename="", storage_queue_limit=50):
self.names_seen = []
self.storage_queue = []
self.storage_queue_limit = storage_queue_limit
self.csv_filename = csv_filename
self.csv_file_open = False
def save_to_csv(self):
try:
self.csv_file_open = True
data_to_save = self.storage_queue.copy()
self.storage_queue.clear()
if not data_to_save:
return
keys = [field.name for field in fields(data_to_save[0])]
# Filter out invalid characters from the filename
valid_chars = "-_.() %s%s" % (string.ascii_letters, string.digits)
valid_filename = ''.join(c for c in self.csv_filename if c in valid_chars)
logger.info(valid_filename)
file_exists = (
os.path.isfile(valid_filename) and os.path.getsize(valid_filename) > 0
)
if not file_exists:
with open(valid_filename, 'w', newline='') as output_file:
writer = csv.DictWriter(output_file, fieldnames=keys)
writer.writeheader()
with open(
valid_filename, mode="a", newline="", encoding="utf-8"
) as output_file:
writer = csv.DictWriter(output_file, fieldnames=keys)
for item in data_to_save:
writer.writerow(asdict(item))
self.csv_file_open = False
except Exception as e:
logger.error(f"Error saving csv {e}")
def is_duplicate(self, input_data):
if input_data.name in self.names_seen:
logger.warning(f"Duplicate item found: {input_data.name}. Item dropped.")
return True
self.names_seen.append(input_data.name)
return False
def add_data(self, scraped_data):
logger.info("adding data")
logger.info(scraped_data)
if not self.is_duplicate(scraped_data):
self.storage_queue.append(scraped_data)
if (
len(self.storage_queue) >= self.storage_queue_limit
and not self.csv_file_open
):
self.save_to_csv()
def close_pipeline(self):
if self.csv_file_open:
time.sleep(3)
if self.storage_queue:
self.save_to_csv()
def scrape_search_results(keyword, page_number, data_pipeline=None, retries=3):
# Use a context manager to ensure the driver is properly closed
with webdriver.Chrome(service=service, options=options) as driver:
formatted_keyword = keyword.replace(" ", "+")
result_number = page_number * 10
logger.info(f"page {page_number}")
url = f"https://www.google.com/search?q={formatted_keyword}%20site%3Aquora.com&start={result_number}"
success = False
tries = 0
while tries < retries and not success:
try:
driver.get(url)
logger.info(f"Accessing {url}")
# Use explicit wait to ensure elements are loaded
wait = WebDriverWait(driver, 10)
wait.until(EC.presence_of_element_located((By.ID, "rso")))
# Extract search result cards
for i in range(1, 11):
try:
# Attempt primary XPath
name = driver.find_element(By.XPATH, f"//*[@id='rso']/div[{i}]/div/div/div[1]/div/div/span/a/h3").text
link = driver.find_element(By.XPATH, f"//*[@id='rso']/div[{i}]/div/div/div[1]/div/div/span/a").get_attribute("href")
except:
try:
# Fallback XPath
name = driver.find_element(By.XPATH, f"//*[@id='rso']/div[{i}]/div/div/div/div[1]/div/div/span/a/h3").text
link = driver.find_element(By.XPATH, f"//*[@id='rso']/div[{i}]/div/div/div/div[1]/div/div/span/a").get_attribute("href")
except Exception as e:
continue
search_data = SearchData(
name=name,
url=link,
rank=result_number + i # Increment rank per result
)
data_pipeline.add_data(search_data)
logger.info(f"Successfully parsed data from: {url}")
success = True
except Exception as e:
logger.error(f"An error occurred while processing {url}: {e}")
tries += 1
if tries >= retries:
logger.error(f"Max retries exceeded for {url}")
else:
logger.info(f"Retrying {url} ({tries}/{retries})")
time.sleep(2)
logger.info(f"Storage queue length after page {page_number}: {len(data_pipeline.storage_queue)}")
def start_scrape(keyword, pages, data_pipeline=None, retries=3):
for page in range(pages):
scrape_search_results(keyword, page, data_pipeline=data_pipeline, retries=retries)
if __name__ == "__main__":
MAX_RETRIES = 3
MAX_THREADS = 5
PAGES = 5
logger.info(f"Crawl starting...")
# INPUT ---> List of keywords to scrape
keyword_list = ["learn rust"]
aggregate_files = []
# Job Processes
for keyword in keyword_list:
filename = keyword.replace(" ", "-")
crawl_pipeline = DataPipeline(csv_filename=f"{filename}.csv")
start_scrape(keyword, PAGES, data_pipeline=crawl_pipeline, max_threads=MAX_THREADS, retries=MAX_RETRIES)
crawl_pipeline.close_pipeline()
aggregate_files.append(f"{filename}.csv")Step 4: Adding Concurrency
Concurrency is essential when scraping large amounts of data. If everything follows a single chain of events, one failure can disrupt the entire process.Concurrency helps distribute tasks across multiple chains that run simultaneously, saving time and improving efficiency.UseThreadPoolExecutor to run concurrent scraping on multiple pages:with ThreadPoolExecutor(max_workers=5) as executor:
executor.submit(scrape_search_results, keyword, page_number)start_scrape function would become:def start_scrape(
keyword, pages, data_pipeline=None, max_threads=5, retries=3
):
with ThreadPoolExecutor(max_workers=max_threads) as executor:
futures = []
for page in range(pages):
# No need to pass the driver anymore, each thread will create its own
futures.append(
executor.submit(
scrape_search_results,
keyword,
page,
data_pipeline,
retries,
)
)
# Ensure all threads complete
for future in futures:
future.result() # This blocks until the thread finishesimport os
import csv
import json
import logging
import time
import string
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.keys import Keys
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, fields, asdict
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Selenium configuration
# Set the path to your ChromeDriver
CHROMEDRIVER_PATH = 'chromedriver.exe' # Adjust this to the actual path if necessary
# Configure the service to use the specified driver
service = Service(CHROMEDRIVER_PATH)
# Setup Chrome options for headless browsing
options = Options()
options.add_argument("--headless")
options.add_argument("--disable-gpu") # Required for headless mode in some environments
options.add_argument("--no-sandbox") # Especially useful for Linux environments
options.add_argument("--disable-dev-shm-usage") # Helps with resource issues on some systems
options.headless = True # Runs Chrome in headless mode (without GUI)
@dataclass
class SearchData:
name: str = ""
url: str = ""
rank: int = 0
def __post_init__(self):
self.check_string_fields()
def check_string_fields(self):
for field in fields(self):
value = getattr(self, field.name)
if isinstance(value, str):
if not value:
setattr(self, field.name, f"No {field.name}")
else:
setattr(self, field.name, value.strip())
class DataPipeline:
def __init__(self, csv_filename="", storage_queue_limit=50):
self.names_seen = []
self.storage_queue = []
self.storage_queue_limit = storage_queue_limit
self.csv_filename = csv_filename
self.csv_file_open = False
def save_to_csv(self):
try:
self.csv_file_open = True
data_to_save = self.storage_queue.copy()
self.storage_queue.clear()
if not data_to_save:
return
keys = [field.name for field in fields(data_to_save[0])]
# Filter out invalid characters from the filename
valid_chars = "-_.() %s%s" % (string.ascii_letters, string.digits)
valid_filename = ''.join(c for c in self.csv_filename if c in valid_chars)
logger.info(valid_filename)
file_exists = (
os.path.isfile(valid_filename) and os.path.getsize(valid_filename) > 0
)
if not file_exists:
with open(valid_filename, 'w', newline='') as output_file:
writer = csv.DictWriter(output_file, fieldnames=keys)
writer.writeheader()
with open(
valid_filename, mode="a", newline="", encoding="utf-8"
) as output_file:
writer = csv.DictWriter(output_file, fieldnames=keys)
for item in data_to_save:
writer.writerow(asdict(item))
self.csv_file_open = False
except Exception as e:
logger.error(f"Error saving csv {e}")
def is_duplicate(self, input_data):
if input_data.name in self.names_seen:
logger.warning(f"Duplicate item found: {input_data.name}. Item dropped.")
return True
self.names_seen.append(input_data.name)
return False
def add_data(self, scraped_data):
logger.info("adding data")
logger.info(scraped_data)
if not self.is_duplicate(scraped_data):
self.storage_queue.append(scraped_data)
if (
len(self.storage_queue) >= self.storage_queue_limit
and not self.csv_file_open
):
self.save_to_csv()
def close_pipeline(self):
if self.csv_file_open:
time.sleep(3)
if self.storage_queue:
self.save_to_csv()
def scrape_search_results(keyword, page_number, data_pipeline=None, retries=3):
# Use a context manager to ensure the driver is properly closed
with webdriver.Chrome(service=service, options=options) as driver:
formatted_keyword = keyword.replace(" ", "+")
result_number = page_number * 10
logger.info(f"page {page_number}")
url = f"https://www.google.com/search?q={formatted_keyword}%20site%3Aquora.com&start={result_number}"
success = False
tries = 0
while tries < retries and not success:
try:
driver.get(url)
logger.info(f"Accessing {url}")
# Use explicit wait to ensure elements are loaded
wait = WebDriverWait(driver, 10)
wait.until(EC.presence_of_element_located((By.ID, "rso")))
# Extract search result cards
for i in range(1, 11):
try:
# Attempt primary XPath
name = driver.find_element(By.XPATH, f"//*[@id='rso']/div[{i}]/div/div/div[1]/div/div/span/a/h3").text
link = driver.find_element(By.XPATH, f"//*[@id='rso']/div[{i}]/div/div/div[1]/div/div/span/a").get_attribute("href")
except:
try:
# Fallback XPath
name = driver.find_element(By.XPATH, f"//*[@id='rso']/div[{i}]/div/div/div/div[1]/div/div/span/a/h3").text
link = driver.find_element(By.XPATH, f"//*[@id='rso']/div[{i}]/div/div/div/div[1]/div/div/span/a").get_attribute("href")
except Exception as e:
continue
search_data = SearchData(
name=name,
url=link,
rank=result_number + i # Increment rank per result
)
data_pipeline.add_data(search_data)
logger.info(f"Successfully parsed data from: {url}")
success = True
except Exception as e:
logger.error(f"An error occurred while processing {url}: {e}")
tries += 1
if tries >= retries:
logger.error(f"Max retries exceeded for {url}")
else:
logger.info(f"Retrying {url} ({tries}/{retries})")
time.sleep(2)
logger.info(f"Storage queue length after page {page_number}: {len(data_pipeline.storage_queue)}")
def start_scrape(
keyword, pages, data_pipeline=None, max_threads=5, retries=3
):
with ThreadPoolExecutor(max_workers=max_threads) as executor:
futures = []
for page in range(pages):
# No need to pass the driver anymore, each thread will create its own
futures.append(
executor.submit(
scrape_search_results,
keyword,
page,
data_pipeline,
retries,
)
)
# Ensure all threads complete
for future in futures:
future.result() # This blocks until the thread finishes
if __name__ == "__main__":
MAX_RETRIES = 3
MAX_THREADS = 5
PAGES = 5
logger.info(f"Crawl starting...")
# INPUT ---> List of keywords to scrape
keyword_list = ["learn rust"]
aggregate_files = []
# Job Processes
for keyword in keyword_list:
filename = keyword.replace(" ", "-")
crawl_pipeline = DataPipeline(csv_filename=f"{filename}.csv")
start_scrape(keyword, PAGES, data_pipeline=crawl_pipeline, max_threads=MAX_THREADS, retries=MAX_RETRIES)
crawl_pipeline.close_pipeline()
aggregate_files.append(f"{filename}.csv")Step 5: Production Run
Once the crawler is complete, setPAGES to the desired number, and initiate a production run. Tweak the following constants as needed:MAX_THREADS = 5 PAGES = 5
if __name__ == "__main__":
MAX_RETRIES = 3
MAX_THREADS = 5
PAGES = 5
LOCATION = "us"
logger.info(f"Crawl starting...")
## INPUT ---> List of keywords to scrape
keyword_list = ["learn rust"]
aggregate_files = []
## Job Processes
for keyword in keyword_list:
filename = keyword.replace(" ", "-")
crawl_pipeline = DataPipeline(csv_filename=f"{filename}.csv")
start_scrape(keyword, PAGES, LOCATION, data_pipeline=crawl_pipeline, max_threads=MAX_THREADS, retries=MAX_RETRIES)
crawl_pipeline.close_pipeline()
aggregate_files.append(f"{filename}.csv")
logger.info(f"Crawl complete.")Screenshot 7The crawling of results from google took 16.469 seconds for 5 pages. 16.469 / 5 = 3.2938 seconds per pageBuild A Quora Scraper
Now to scrape the answers from quora, following steps would be taken:- Read the csv file generated from google search results.
- Open each post from the search results in the csv files and parse the answers data from the posts pages.
- Store the data.
- Perform concurrency for step 2 and 3 to proccess multiple posts at the same time.
- Run the Scraper
Step 1: Create Simple Business Data Parser
The goal of this step is to scrape each Quora post and extract the main content, specifically the answers and relevant replies, while filtering out non-relevant data such as promoted or related responses.Theprocess_post function is responsible for visiting a Quora post, waiting for the content to load, and then extracting the answers. It uses Selenium to interact with the dynamically loaded elements on the Quora page.Here's how it works:def process_post(row, retries=3):
with webdriver.Chrome(service=service, options=options) as driver:
logger.info(f"Processing row: {row}")
url = row.get("url")
if not url:
logger.error(f"No URL found in row: {row}")
return
success = False
tries = 0
while tries < retries and not success:
try:
# Step 1: Open the URL and wait for the main content to load
driver.get(url)
logger.info(f"Accessing {url}")
wait = WebDriverWait(driver, 10)
main_content = wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, "div[id='mainContent']")))
# Step 2: Extract answer cards
answer_cards = main_content.find_elements(By.CSS_SELECTOR, "div.q-click-wrapper")
if not answer_cards:
logger.warning(f"No answer cards found at {url}")
# Step 3: Initialize a DataPipeline to store replies
if 'name' not in row:
logger.error(f"'name' key missing in row: {row}")
break
last_seen_name = ""
# Step 4: Loop through each answer card and extract name and reply
for answer_card in answer_cards:
try:
name_element = answer_card.find_element(By.CSS_SELECTOR, "div.q-relative")
name = name_element.text.replace("\n", "").strip()
reply_element = answer_card.find_element(By.CSS_SELECTOR, "div.spacing_log_answer_content")
reply = reply_element.text.strip()
# Filter out promoted content and related questions
if "Sponsored" in name:
continue
if "Related questions" in name:
break
if name == last_seen_name:
continue
last_seen_name = name
print("name:", name)
print("reply:", reply)
except Exception as e:
continue
success = True
except Exception as e:
logger.error(f"Exception thrown while processing {url}: {e}")
tries += 1
if tries >= retries:
logger.error(f"Max retries exceeded for {url}")
else:
logger.info(f"Retrying {url} ({tries}/{retries})")
time.sleep(2)- Extracting Answers: Uses Selenium to wait for and locate the main content of a Quora post. It then extracts each individual answer (skipping promoted or irrelevant content) and stores it.
- Retries: Includes a retry mechanism to handle temporary failures, such as page load errors.
Step 2: Loading URLs To Scrape
Once you have scraped URLs from Google search results, you need to load these URLs into the scraper to process each Quora post.This is handled by theprocess_results function, which loads URLs from a CSV file and calls process_post to scrape each post.def process_results(csv_file, max_threads=5, retries=3):
with open(csv_file, newline="") as file:
reader = list(csv.DictReader(file))
logger.info(f"file opened")
for row in reader:
process_post(row, retries)- Reading URLs: This function opens the CSV file generated by the search crawler, reads the URLs of the Quora posts, and processes them sequentially.
- Processing Each Post: For each URL, it calls the
process_postfunction to scrape the content.
import os
import csv
import json
import logging
import time
import string
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.keys import Keys
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, fields, asdict
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Selenium configuration
# Set the path to your ChromeDriver
CHROMEDRIVER_PATH = 'chromedriver.exe' # Adjust this to the actual path if necessary
# Configure the service to use the specified driver
service = Service(CHROMEDRIVER_PATH)
# Setup Chrome options for headless browsing
options = Options()
options.add_argument("--headless")
options.add_argument("--disable-gpu") # Required for headless mode in some environments
options.add_argument("--no-sandbox") # Especially useful for Linux environments
options.add_argument("--disable-dev-shm-usage") # Helps with resource issues on some systems
options.headless = True # Runs Chrome in headless mode (without GUI)
@dataclass
class SearchData:
name: str = ""
url: str = ""
rank: int = 0
def __post_init__(self):
self.check_string_fields()
def check_string_fields(self):
for field in fields(self):
value = getattr(self, field.name)
if isinstance(value, str):
if not value:
setattr(self, field.name, f"No {field.name}")
else:
setattr(self, field.name, value.strip())
class DataPipeline:
def __init__(self, csv_filename="", storage_queue_limit=50):
self.names_seen = []
self.storage_queue = []
self.storage_queue_limit = storage_queue_limit
self.csv_filename = csv_filename
self.csv_file_open = False
def save_to_csv(self):
try:
self.csv_file_open = True
data_to_save = self.storage_queue.copy()
self.storage_queue.clear()
if not data_to_save:
return
keys = [field.name for field in fields(data_to_save[0])]
# Filter out invalid characters from the filename
valid_chars = "-_.() %s%s" % (string.ascii_letters, string.digits)
valid_filename = ''.join(c for c in self.csv_filename if c in valid_chars)
logger.info(valid_filename)
file_exists = (
os.path.isfile(valid_filename) and os.path.getsize(valid_filename) > 0
)
if not file_exists:
with open(valid_filename, 'w', newline='') as output_file:
writer = csv.DictWriter(output_file, fieldnames=keys)
writer.writeheader()
with open(
valid_filename, mode="a", newline="", encoding="utf-8"
) as output_file:
writer = csv.DictWriter(output_file, fieldnames=keys)
for item in data_to_save:
writer.writerow(asdict(item))
self.csv_file_open = False
except Exception as e:
logger.error(f"Error saving csv {e}")
def is_duplicate(self, input_data):
if input_data.name in self.names_seen:
logger.warning(f"Duplicate item found: {input_data.name}. Item dropped.")
return True
self.names_seen.append(input_data.name)
return False
def add_data(self, scraped_data):
logger.info("adding data")
logger.info(scraped_data)
if not self.is_duplicate(scraped_data):
self.storage_queue.append(scraped_data)
if (
len(self.storage_queue) >= self.storage_queue_limit
and not self.csv_file_open
):
self.save_to_csv()
def close_pipeline(self):
if self.csv_file_open:
time.sleep(3)
if self.storage_queue:
self.save_to_csv()
def scrape_search_results(keyword, page_number, data_pipeline=None, retries=3):
# Use a context manager to ensure the driver is properly closed
with webdriver.Chrome(service=service, options=options) as driver:
formatted_keyword = keyword.replace(" ", "+")
result_number = page_number * 10
logger.info(f"page {page_number}")
url = f"https://www.google.com/search?q={formatted_keyword}%20site%3Aquora.com&start={result_number}"
success = False
tries = 0
while tries < retries and not success:
try:
driver.get(url)
logger.info(f"Accessing {url}")
# Use explicit wait to ensure elements are loaded
wait = WebDriverWait(driver, 10)
wait.until(EC.presence_of_element_located((By.ID, "rso")))
# Extract search result cards
for i in range(1, 11):
try:
# Attempt primary XPath
name = driver.find_element(By.XPATH, f"//*[@id='rso']/div[{i}]/div/div/div[1]/div/div/span/a/h3").text
link = driver.find_element(By.XPATH, f"//*[@id='rso']/div[{i}]/div/div/div[1]/div/div/span/a").get_attribute("href")
except:
try:
# Fallback XPath
name = driver.find_element(By.XPATH, f"//*[@id='rso']/div[{i}]/div/div/div/div[1]/div/div/span/a/h3").text
link = driver.find_element(By.XPATH, f"//*[@id='rso']/div[{i}]/div/div/div/div[1]/div/div/span/a").get_attribute("href")
except Exception as e:
continue
search_data = SearchData(
name=name,
url=link,
rank=result_number + i # Increment rank per result
)
data_pipeline.add_data(search_data)
logger.info(f"Successfully parsed data from: {url}")
success = True
except Exception as e:
logger.error(f"An error occurred while processing {url}: {e}")
tries += 1
if tries >= retries:
logger.error(f"Max retries exceeded for {url}")
else:
logger.info(f"Retrying {url} ({tries}/{retries})")
time.sleep(2)
logger.info(f"Storage queue length after page {page_number}: {len(data_pipeline.storage_queue)}")
def start_scrape(
keyword, pages, data_pipeline=None, max_threads=5, retries=3
):
with ThreadPoolExecutor(max_workers=max_threads) as executor:
futures = []
for page in range(pages):
# No need to pass the driver anymore, each thread will create its own
futures.append(
executor.submit(
scrape_search_results,
keyword,
page,
data_pipeline,
retries,
)
)
# Ensure all threads complete
for future in futures:
future.result() # This blocks until the thread finishes
def process_post(row, retries=3):
with webdriver.Chrome(service=service, options=options) as driver:
logger.info(f"Processing row: {row}")
url = row.get("url")
if not url:
logger.error(f"No URL found in row: {row}")
return
success = False
tries = 0
while tries < retries and not success:
try:
# Step 1: Open the URL and wait for the main content to load
driver.get(url)
logger.info(f"Accessing {url}")
wait = WebDriverWait(driver, 10)
main_content = wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, "div[id='mainContent']")))
# Step 2: Extract answer cards
answer_cards = main_content.find_elements(By.CSS_SELECTOR, "div.q-click-wrapper")
if not answer_cards:
logger.warning(f"No answer cards found at {url}")
# Step 3: Initialize a DataPipeline to store replies
if 'name' not in row:
logger.error(f"'name' key missing in row: {row}")
break
last_seen_name = ""
# Step 4: Loop through each answer card and extract name and reply
for answer_card in answer_cards:
try:
name_element = answer_card.find_element(By.CSS_SELECTOR, "div.q-relative")
name = name_element.text.replace("\n", "").strip()
reply_element = answer_card.find_element(By.CSS_SELECTOR, "div.spacing_log_answer_content")
reply = reply_element.text.strip()
# Filter out promoted content and related questions
if "Sponsored" in name:
continue
if "Related questions" in name:
break
if name == last_seen_name:
continue
last_seen_name = name
print("name:", name)
print("reply:", reply)
except Exception as e:
continue
success = True
except Exception as e:
logger.error(f"Exception thrown while processing {url}: {e}")
tries += 1
if tries >= retries:
logger.error(f"Max retries exceeded for {url}")
else:
logger.info(f"Retrying {url} ({tries}/{retries})")
time.sleep(2)
def process_results(csv_file, max_threads=5, retries=3):
with open(csv_file, newline="") as file:
reader = list(csv.DictReader(file))
logger.info(f"file opened")
for row in reader:
process_post(row, retries)
if __name__ == "__main__":
MAX_RETRIES = 3
MAX_THREADS = 5
PAGES = 5
logger.info(f"Crawl starting...")
# INPUT ---> List of keywords to scrape
keyword_list = ["learn rust"]
aggregate_files = []
# Job Processes
for keyword in keyword_list:
filename = keyword.replace(" ", "-")
crawl_pipeline = DataPipeline(csv_filename=f"{filename}.csv")
start_scrape(keyword, PAGES, data_pipeline=crawl_pipeline, max_threads=MAX_THREADS, retries=MAX_RETRIES)
crawl_pipeline.close_pipeline()
aggregate_files.append(f"{filename}.csv")
logger.info(f"Crawl complete.")
for file in aggregate_files:
process_results(file, max_threads=MAX_THREADS, retries=MAX_RETRIES)Step 3: Storing the Scraped Data
The extracted data (such as answers and user names) is stored in a CSV file using theReplyData class and the DataPipeline class.Each scraped answer is stored as an instance of ReplyData, ensuring that the scraped content is well-structured.ReplyData Class:
@dataclass
class ReplyData:
name: str = ""
reply: str = ""
def __post_init__(self):
self.check_string_fields()
def check_string_fields(self):
for field in fields(self):
value = getattr(self, field.name)
if isinstance(value, str):
if not value:
setattr(self, field.name, f"No {field.name}")
else:
setattr(self, field.name, value.strip())- Data Structuring: The
ReplyDataclass is used to store the name of the user and the content of their reply in a structured format. - Field Validation: The
check_string_fieldsmethod ensures that empty or malformed strings are handled by assigning a default value or removing unnecessary whitespace.
ReplyData is passed to the DataPipeline class for storage in a CSV file.The full code would be:import os
import csv
import json
import logging
import time
import string
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.keys import Keys
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, fields, asdict
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Selenium configuration
# Set the path to your ChromeDriver
CHROMEDRIVER_PATH = 'chromedriver.exe' # Adjust this to the actual path if necessary
# Configure the service to use the specified driver
service = Service(CHROMEDRIVER_PATH)
# Setup Chrome options for headless browsing
options = Options()
options.add_argument("--headless")
options.add_argument("--disable-gpu") # Required for headless mode in some environments
options.add_argument("--no-sandbox") # Especially useful for Linux environments
options.add_argument("--disable-dev-shm-usage") # Helps with resource issues on some systems
options.headless = True # Runs Chrome in headless mode (without GUI)
@dataclass
class SearchData:
name: str = ""
url: str = ""
rank: int = 0
def __post_init__(self):
self.check_string_fields()
def check_string_fields(self):
for field in fields(self):
value = getattr(self, field.name)
if isinstance(value, str):
if not value:
setattr(self, field.name, f"No {field.name}")
else:
setattr(self, field.name, value.strip())
@dataclass
class ReplyData:
name: str = ""
reply: str = ""
def __post_init__(self):
self.check_string_fields()
def check_string_fields(self):
for field in fields(self):
value = getattr(self, field.name)
if isinstance(value, str):
if not value:
setattr(self, field.name, f"No {field.name}")
else:
setattr(self, field.name, value.strip())
class DataPipeline:
def __init__(self, csv_filename="", storage_queue_limit=50):
self.names_seen = []
self.storage_queue = []
self.storage_queue_limit = storage_queue_limit
self.csv_filename = csv_filename
self.csv_file_open = False
def save_to_csv(self):
try:
self.csv_file_open = True
data_to_save = self.storage_queue.copy()
self.storage_queue.clear()
if not data_to_save:
return
keys = [field.name for field in fields(data_to_save[0])]
# Filter out invalid characters from the filename
valid_chars = "-_.() %s%s" % (string.ascii_letters, string.digits)
valid_filename = ''.join(c for c in self.csv_filename if c in valid_chars)
logger.info(valid_filename)
file_exists = (
os.path.isfile(valid_filename) and os.path.getsize(valid_filename) > 0
)
if not file_exists:
with open(valid_filename, 'w', newline='') as output_file:
writer = csv.DictWriter(output_file, fieldnames=keys)
writer.writeheader()
with open(
valid_filename, mode="a", newline="", encoding="utf-8"
) as output_file:
writer = csv.DictWriter(output_file, fieldnames=keys)
for item in data_to_save:
writer.writerow(asdict(item))
self.csv_file_open = False
except Exception as e:
logger.error(f"Error saving csv {e}")
def is_duplicate(self, input_data):
if input_data.name in self.names_seen:
logger.warning(f"Duplicate item found: {input_data.name}. Item dropped.")
return True
self.names_seen.append(input_data.name)
return False
def add_data(self, scraped_data):
logger.info("adding data")
logger.info(scraped_data)
if not self.is_duplicate(scraped_data):
self.storage_queue.append(scraped_data)
if (
len(self.storage_queue) >= self.storage_queue_limit
and not self.csv_file_open
):
self.save_to_csv()
def close_pipeline(self):
if self.csv_file_open:
time.sleep(3)
if self.storage_queue:
self.save_to_csv()
def scrape_search_results(keyword, page_number, data_pipeline=None, retries=3):
# Use a context manager to ensure the driver is properly closed
with webdriver.Chrome(service=service, options=options) as driver:
formatted_keyword = keyword.replace(" ", "+")
result_number = page_number * 10
logger.info(f"page {page_number}")
url = f"https://www.google.com/search?q={formatted_keyword}%20site%3Aquora.com&start={result_number}"
success = False
tries = 0
while tries < retries and not success:
try:
driver.get(url)
logger.info(f"Accessing {url}")
# Use explicit wait to ensure elements are loaded
wait = WebDriverWait(driver, 10)
wait.until(EC.presence_of_element_located((By.ID, "rso")))
# Extract search result cards
for i in range(1, 11):
try:
# Attempt primary XPath
name = driver.find_element(By.XPATH, f"//*[@id='rso']/div[{i}]/div/div/div[1]/div/div/span/a/h3").text
link = driver.find_element(By.XPATH, f"//*[@id='rso']/div[{i}]/div/div/div[1]/div/div/span/a").get_attribute("href")
except:
try:
# Fallback XPath
name = driver.find_element(By.XPATH, f"//*[@id='rso']/div[{i}]/div/div/div/div[1]/div/div/span/a/h3").text
link = driver.find_element(By.XPATH, f"//*[@id='rso']/div[{i}]/div/div/div/div[1]/div/div/span/a").get_attribute("href")
except Exception as e:
continue
search_data = SearchData(
name=name,
url=link,
rank=result_number + i # Increment rank per result
)
data_pipeline.add_data(search_data)
logger.info(f"Successfully parsed data from: {url}")
success = True
except Exception as e:
logger.error(f"An error occurred while processing {url}: {e}")
tries += 1
if tries >= retries:
logger.error(f"Max retries exceeded for {url}")
else:
logger.info(f"Retrying {url} ({tries}/{retries})")
time.sleep(2)
logger.info(f"Storage queue length after page {page_number}: {len(data_pipeline.storage_queue)}")
def start_scrape(
keyword, pages, data_pipeline=None, max_threads=5, retries=3
):
with ThreadPoolExecutor(max_workers=max_threads) as executor:
futures = []
for page in range(pages):
# No need to pass the driver anymore, each thread will create its own
futures.append(
executor.submit(
scrape_search_results,
keyword,
page,
data_pipeline,
retries,
)
)
# Ensure all threads complete
for future in futures:
future.result() # This blocks until the thread finishes
def process_post(row, retries=3):
with webdriver.Chrome(service=service, options=options) as driver:
logger.info(f"Processing row: {row}")
url = row.get("url")
if not url:
logger.error(f"No URL found in row: {row}")
return
logger.info(f"Processing URL: {url}")
success = False
tries = 0
while tries < retries and not success:
try:
driver.get(url)
logger.info(f"Accessing {url}")
# Use explicit wait to ensure main content is loaded
wait = WebDriverWait(driver, 10)
main_content = wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, "div[id='mainContent']")))
# Extract answer cards
answer_cards = main_content.find_elements(By.CSS_SELECTOR, "div.q-click-wrapper")
if not answer_cards:
logger.warning(f"No answer cards found at {url}")
# Initialize a new DataPipeline for replies
if 'name' not in row:
logger.error(f"'name' key missing in row: {row}")
break
answer_pipeline = DataPipeline(
csv_filename=f"{row['name'].replace(' ', '-')}.csv"
)
last_seen_name = ""
for answer_card in answer_cards:
try:
name_element = answer_card.find_element(By.CSS_SELECTOR, "div.q-relative")
name = name_element.text.replace("\n", "").strip()
reply_element = answer_card.find_element(By.CSS_SELECTOR, "div.spacing_log_answer_content")
reply = reply_element.text.strip()
if "Sponsored" in name:
continue
if "Related questions" in name:
break
if name == last_seen_name:
continue
last_seen_name = name
reply_data = ReplyData(name=name, reply=reply)
answer_pipeline.add_data(reply_data)
except Exception as e:
continue
answer_pipeline.close_pipeline()
success = True
except Exception as e:
logger.error(f"Exception thrown while processing {url}: {e}")
tries += 1
if tries >= retries:
logger.error(f"Max retries exceeded for {url}")
else:
logger.info(f"Retrying {url} ({tries}/{retries})")
time.sleep(2)
def process_results(csv_file, max_threads=5, retries=3):
with open(csv_file, newline="") as file:
reader = list(csv.DictReader(file))
logger.info(f"file opened")
for row in reader:
process_post(row, retries)
if __name__ == "__main__":
MAX_RETRIES = 3
MAX_THREADS = 5
PAGES = 5
logger.info(f"Crawl starting...")
# INPUT ---> List of keywords to scrape
keyword_list = ["learn rust"]
aggregate_files = []
# Job Processes
for keyword in keyword_list:
filename = keyword.replace(" ", "-")
crawl_pipeline = DataPipeline(csv_filename=f"{filename}.csv")
start_scrape(keyword, PAGES, data_pipeline=crawl_pipeline, max_threads=MAX_THREADS, retries=MAX_RETRIES)
crawl_pipeline.close_pipeline()
aggregate_files.append(f"{filename}.csv")
logger.info(f"Crawl complete.")
for file in aggregate_files:
process_results(file, max_threads=MAX_THREADS, retries=MAX_RETRIES)Step 4: Adding Concurrency
To scrape multiple Quora posts concurrently and improve efficiency, you can modify theprocess_results function to use ThreadPoolExecutor. This allows the scraper to handle multiple posts at once, significantly speeding up the process.from concurrent.futures import ThreadPoolExecutor
def process_results(csv_file, max_threads=5, retries=3):
with open(csv_file, newline="") as file:
reader = list(csv.DictReader(file))
logger.info(f"file opened")
with ThreadPoolExecutor(max_workers=max_threads) as executor:
for row in reader:
executor.submit(process_post, row, retries)- Threading:
ThreadPoolExecutoris used to run multiple threads, allowing the scraper to process several Quora posts simultaneously. - Concurrency: The
max_workersparameter defines the number of threads running concurrently. Each thread calls theprocess_postfunction to handle a single Quora post.
import os
import csv
import json
import logging
import time
import string
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.keys import Keys
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, fields, asdict
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Selenium configuration
# Set the path to your ChromeDriver
CHROMEDRIVER_PATH = 'chromedriver.exe' # Adjust this to the actual path if necessary
# Configure the service to use the specified driver
service = Service(CHROMEDRIVER_PATH)
# Setup Chrome options for headless browsing
options = Options()
options.add_argument("--headless")
options.add_argument("--disable-gpu") # Required for headless mode in some environments
options.add_argument("--no-sandbox") # Especially useful for Linux environments
options.add_argument("--disable-dev-shm-usage") # Helps with resource issues on some systems
options.headless = True # Runs Chrome in headless mode (without GUI)
@dataclass
class SearchData:
name: str = ""
url: str = ""
rank: int = 0
def __post_init__(self):
self.check_string_fields()
def check_string_fields(self):
for field in fields(self):
value = getattr(self, field.name)
if isinstance(value, str):
if not value:
setattr(self, field.name, f"No {field.name}")
else:
setattr(self, field.name, value.strip())
@dataclass
class ReplyData:
name: str = ""
reply: str = ""
def __post_init__(self):
self.check_string_fields()
def check_string_fields(self):
for field in fields(self):
value = getattr(self, field.name)
if isinstance(value, str):
if not value:
setattr(self, field.name, f"No {field.name}")
else:
setattr(self, field.name, value.strip())
class DataPipeline:
def __init__(self, csv_filename="", storage_queue_limit=50):
self.names_seen = []
self.storage_queue = []
self.storage_queue_limit = storage_queue_limit
self.csv_filename = csv_filename
self.csv_file_open = False
def save_to_csv(self):
try:
self.csv_file_open = True
data_to_save = self.storage_queue.copy()
self.storage_queue.clear()
if not data_to_save:
return
keys = [field.name for field in fields(data_to_save[0])]
# Filter out invalid characters from the filename
valid_chars = "-_.() %s%s" % (string.ascii_letters, string.digits)
valid_filename = ''.join(c for c in self.csv_filename if c in valid_chars)
logger.info(valid_filename)
file_exists = (
os.path.isfile(valid_filename) and os.path.getsize(valid_filename) > 0
)
if not file_exists:
with open(valid_filename, 'w', newline='') as output_file:
writer = csv.DictWriter(output_file, fieldnames=keys)
writer.writeheader()
with open(
valid_filename, mode="a", newline="", encoding="utf-8"
) as output_file:
writer = csv.DictWriter(output_file, fieldnames=keys)
for item in data_to_save:
writer.writerow(asdict(item))
self.csv_file_open = False
except Exception as e:
logger.error(f"Error saving csv {e}")
def is_duplicate(self, input_data):
if input_data.name in self.names_seen:
logger.warning(f"Duplicate item found: {input_data.name}. Item dropped.")
return True
self.names_seen.append(input_data.name)
return False
def add_data(self, scraped_data):
logger.info("adding data")
logger.info(scraped_data)
if not self.is_duplicate(scraped_data):
self.storage_queue.append(scraped_data)
if (
len(self.storage_queue) >= self.storage_queue_limit
and not self.csv_file_open
):
self.save_to_csv()
def close_pipeline(self):
if self.csv_file_open:
time.sleep(3)
if self.storage_queue:
self.save_to_csv()
def scrape_search_results(keyword, page_number, data_pipeline=None, retries=3):
# Use a context manager to ensure the driver is properly closed
with webdriver.Chrome(service=service, options=options) as driver:
formatted_keyword = keyword.replace(" ", "+")
result_number = page_number * 10
logger.info(f"page {page_number}")
url = f"https://www.google.com/search?q={formatted_keyword}%20site%3Aquora.com&start={result_number}"
success = False
tries = 0
while tries < retries and not success:
try:
driver.get(url)
logger.info(f"Accessing {url}")
# Use explicit wait to ensure elements are loaded
wait = WebDriverWait(driver, 10)
wait.until(EC.presence_of_element_located((By.ID, "rso")))
# Extract search result cards
for i in range(1, 11):
try:
# Attempt primary XPath
name = driver.find_element(By.XPATH, f"//*[@id='rso']/div[{i}]/div/div/div[1]/div/div/span/a/h3").text
link = driver.find_element(By.XPATH, f"//*[@id='rso']/div[{i}]/div/div/div[1]/div/div/span/a").get_attribute("href")
except:
try:
# Fallback XPath
name = driver.find_element(By.XPATH, f"//*[@id='rso']/div[{i}]/div/div/div/div[1]/div/div/span/a/h3").text
link = driver.find_element(By.XPATH, f"//*[@id='rso']/div[{i}]/div/div/div/div[1]/div/div/span/a").get_attribute("href")
except Exception as e:
continue
search_data = SearchData(
name=name,
url=link,
rank=result_number + i # Increment rank per result
)
data_pipeline.add_data(search_data)
logger.info(f"Successfully parsed data from: {url}")
success = True
except Exception as e:
logger.error(f"An error occurred while processing {url}: {e}")
tries += 1
if tries >= retries:
logger.error(f"Max retries exceeded for {url}")
else:
logger.info(f"Retrying {url} ({tries}/{retries})")
time.sleep(2)
logger.info(f"Storage queue length after page {page_number}: {len(data_pipeline.storage_queue)}")
def start_scrape(
keyword, pages, data_pipeline=None, max_threads=5, retries=3
):
with ThreadPoolExecutor(max_workers=max_threads) as executor:
futures = []
for page in range(pages):
# No need to pass the driver anymore, each thread will create its own
futures.append(
executor.submit(
scrape_search_results,
keyword,
page,
data_pipeline,
retries,
)
)
# Ensure all threads complete
for future in futures:
future.result() # This blocks until the thread finishes
def process_post(row, retries=3):
with webdriver.Chrome(service=service, options=options) as driver:
logger.info(f"Processing row: {row}")
url = row.get("url")
if not url:
logger.error(f"No URL found in row: {row}")
return
logger.info(f"Processing URL: {url}")
success = False
tries = 0
while tries < retries and not success:
try:
driver.get(url)
logger.info(f"Accessing {url}")
# Use explicit wait to ensure main content is loaded
wait = WebDriverWait(driver, 10)
main_content = wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, "div[id='mainContent']")))
# Extract answer cards
answer_cards = main_content.find_elements(By.CSS_SELECTOR, "div.q-click-wrapper")
if not answer_cards:
logger.warning(f"No answer cards found at {url}")
# Initialize a new DataPipeline for replies
if 'name' not in row:
logger.error(f"'name' key missing in row: {row}")
break
answer_pipeline = DataPipeline(
csv_filename=f"{row['name'].replace(' ', '-')}.csv"
)
last_seen_name = ""
for answer_card in answer_cards:
try:
name_element = answer_card.find_element(By.CSS_SELECTOR, "div.q-relative")
name = name_element.text.replace("\n", "").strip()
reply_element = answer_card.find_element(By.CSS_SELECTOR, "div.spacing_log_answer_content")
reply = reply_element.text.strip()
if "Sponsored" in name:
continue
if "Related questions" in name:
break
if name == last_seen_name:
continue
last_seen_name = name
reply_data = ReplyData(name=name, reply=reply)
answer_pipeline.add_data(reply_data)
except Exception as e:
continue
answer_pipeline.close_pipeline()
success = True
except Exception as e:
logger.error(f"Exception thrown while processing {url}: {e}")
tries += 1
if tries >= retries:
logger.error(f"Max retries exceeded for {url}")
else:
logger.info(f"Retrying {url} ({tries}/{retries})")
time.sleep(2)
def process_results(csv_file, max_threads=5, retries=3):
with open(csv_file, newline="") as file:
reader = list(csv.DictReader(file))
logger.info(f"file opened")
with ThreadPoolExecutor(max_workers=max_threads) as executor:
for row in reader:
executor.submit(process_post, row, retries)
if __name__ == "__main__":
MAX_RETRIES = 3
MAX_THREADS = 5
PAGES = 5
logger.info(f"Crawl starting...")
# INPUT ---> List of keywords to scrape
keyword_list = ["learn rust"]
aggregate_files = []
# Job Processes
for keyword in keyword_list:
filename = keyword.replace(" ", "-")
crawl_pipeline = DataPipeline(csv_filename=f"{filename}.csv")
start_scrape(keyword, PAGES, data_pipeline=crawl_pipeline, max_threads=MAX_THREADS, retries=MAX_RETRIES)
crawl_pipeline.close_pipeline()
aggregate_files.append(f"{filename}.csv")
logger.info(f"Crawl complete.")
for file in aggregate_files:
process_results(file, max_threads=MAX_THREADS, retries=MAX_RETRIES)Step 5: Production Run
Finally, when your scraper is ready to run on a larger dataset, you can execute the full scraping process. This includes scraping search results for multiple keywords, storing the data in CSV files, and processing the scraped URLs concurrently.Themain function would be:if __name__ == "__main__":
MAX_RETRIES = 3
MAX_THREADS = 5
PAGES = 5
logger.info(f"Crawl starting...")
# INPUT ---> List of keywords to scrape
keyword_list = ["learn rust"]
aggregate_files = []
# Job Processes: Scraping Search Results
for keyword in keyword_list:
filename = keyword.replace(" ", "-")
crawl_pipeline = DataPipeline(csv_filename=f"{filename}.csv")
start_scrape(keyword, PAGES, data_pipeline=crawl_pipeline, max_threads=MAX_THREADS, retries=MAX_RETRIES)
crawl_pipeline.close_pipeline()
aggregate_files.append(f"{filename}.csv")
logger.info(f"Crawl complete.")
# Processing Scraped Quora Posts
for file in aggregate_files:
process_results(file, max_threads=MAX_THREADS, retries=MAX_RETRIES)- Scraping Search Results: For each keyword in the
keyword_list, the scraper collects search results from Google and stores them in a CSV file. - Concurrent Processing: After collecting the URLs, the
process_resultsfunction processes each Quora post concurrently, using multiple threads for efficiency. - Parameters: You can adjust
MAX_RETRIES,MAX_THREADS, andPAGESto fine-tune the scraper's performance. More threads will increase speed, but be mindful of server load and anti-bot measures.
Screeshot 8The crawling of all the quora posts took 651.258 seconds and out of those 16.469 were taken by google. So, the scraping of posts took: 651.258 - 16.469 = 634.789 seconds. We scraped 50 posts so, 634.789 / 50 = 12.695 seconds per pageLegal and Ethical Considerations
When scraping the web, you need to pay attention to your target site's Terms of Service and theirrobots.txt. Legal or not, when you violate a site's terms, you can get suspended or even permanently banned.Public data is typically free to scrape, but be cautious when dealing with private or gated content.When scraping Quora, be mindful of their Terms of Service and review their robots.txt file.Ensure that your scraping activities do not violate legal or ethical guidelines.When scraping private data, you are subject to a site's terms and privacy laws in the site's jurisdiction. If you don't know if your scraper is legal, you should consult an attorney.