made and tested code to scrape twitter for a specific topic
This commit is contained in:
simplypower-bbj
+87
@@ -0,0 +1,87 @@
|
||||
from selenium.webdriver.common.action_chains import ActionChains
|
||||
from dateutil.parser import parse
|
||||
from selenium import webdriver
|
||||
from selenium.webdriver.common.by import By
|
||||
from selenium.webdriver.support import expected_conditions as EC
|
||||
from selenium.webdriver.support.ui import WebDriverWait
|
||||
from selenium.webdriver.chrome.options import Options
|
||||
from fake_useragent import UserAgent
|
||||
|
||||
def load_topic(topic, headless=True):
|
||||
# instantiate driver
|
||||
options = Options()
|
||||
if headless:
|
||||
options.headless = True
|
||||
# fake a user agent to not be denied service
|
||||
ua = UserAgent()
|
||||
userAgent = ua.random
|
||||
options.add_argument(f'user-agent={userAgent}')
|
||||
driver = webdriver.Chrome(options=options)
|
||||
# open webpage
|
||||
url = f'https://twitter.com/search?q=%23{topic}&src=typed_query&f=live'
|
||||
driver.get(url)
|
||||
# wait for popup and click it
|
||||
timeout = 30 # seconds
|
||||
xpath_str = "//*[text()='Not now']"
|
||||
element = WebDriverWait(driver, timeout).until(EC.presence_of_element_located((By.XPATH, xpath_str)))
|
||||
element.click()
|
||||
return driver
|
||||
|
||||
def list_tweets(driver):
|
||||
timeout = 30 # seconds
|
||||
_ = WebDriverWait(driver, timeout).until(EC.presence_of_element_located((By.CSS_SELECTOR, '[data-testid="tweet"]')))
|
||||
return driver.find_elements(By.CSS_SELECTOR, '[data-testid="tweet"]')
|
||||
|
||||
def scroll_to(driver, tweet):
|
||||
action = ActionChains(driver)
|
||||
action.move_to_element(tweet).perform()
|
||||
|
||||
def get_id(tweet):
|
||||
return tweet.id
|
||||
|
||||
def get_timestamp(tweet):
|
||||
timestamp = tweet.find_element(By.TAG_NAME, "time").get_attribute("datetime")
|
||||
return parse(timestamp)
|
||||
|
||||
def get_content(tweet):
|
||||
return tweet.find_element(By.CSS_SELECTOR, 'div[lang]').text
|
||||
|
||||
def get_username(tweet):
|
||||
return tweet.find_element(By.XPATH, ".//span[contains(text(), '@')]").text
|
||||
|
||||
def extract(driver, limit=100):
|
||||
id_list = list()
|
||||
username_list = list()
|
||||
timestamp_list = list()
|
||||
content_list = list()
|
||||
while len(id_list) < limit:
|
||||
tweet_list = list_tweets(driver)
|
||||
for tweet in tweet_list:
|
||||
id = get_id(tweet)
|
||||
if id in id_list: continue
|
||||
scroll_to(driver, tweet)
|
||||
username_list.append(get_username(tweet))
|
||||
timestamp_list.append(get_timestamp(tweet))
|
||||
content_list.append(get_content(tweet))
|
||||
id_list.append(id)
|
||||
return username_list, timestamp_list, content_list
|
||||
|
||||
def scrape(topic, limit=100, headless=True):
|
||||
try:
|
||||
# prepare driver
|
||||
driver = load_topic(topic, headless)
|
||||
except Exception as e:
|
||||
print('failed instantiating webdriver')
|
||||
print(e)
|
||||
return [], [], []
|
||||
try:
|
||||
# extract information
|
||||
username_list, timestamp_list, content_list = extract(driver, limit)
|
||||
except Exception as e:
|
||||
print('error when extracting tweets')
|
||||
print(e)
|
||||
finally:
|
||||
# close driver
|
||||
driver.close()
|
||||
driver.quit()
|
||||
return username_list, timestamp_list, content_list
|
||||
+96
@@ -0,0 +1,96 @@
|
||||
{
|
||||
"cells": [
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 12,
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"name": "stdout",
|
||||
"output_type": "stream",
|
||||
"text": [
|
||||
"12\n"
|
||||
]
|
||||
}
|
||||
],
|
||||
"source": [
|
||||
"from selenium.webdriver.common.action_chains import ActionChains\n",
|
||||
"from dateutil.parser import parse\n",
|
||||
"from selenium import webdriver\n",
|
||||
"import os\n",
|
||||
"import importlib\n",
|
||||
"import scraper\n",
|
||||
"\n",
|
||||
"importlib.reload(scraper)\n",
|
||||
"topic = 'Odense'\n",
|
||||
"limit = 10\n",
|
||||
"username_list, timestamp_list, content_list = scraper.scrape(topic, limit, headless=False)\n",
|
||||
"\n",
|
||||
"print(len(username_list))"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 11,
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"ename": "TimeoutException",
|
||||
"evalue": "Message: \nStacktrace:\n0 chromedriver 0x000000010546ff38 chromedriver + 4910904\n1 chromedriver 0x00000001053efa03 chromedriver + 4385283\n2 chromedriver 0x0000000105034747 chromedriver + 472903\n3 chromedriver 0x000000010507934c chromedriver + 754508\n4 chromedriver 0x00000001050795a1 chromedriver + 755105\n5 chromedriver 0x00000001050bce94 chromedriver + 1031828\n6 chromedriver 0x000000010509f13d chromedriver + 909629\n7 chromedriver 0x00000001050ba28e chromedriver + 1020558\n8 chromedriver 0x000000010509eee3 chromedriver + 909027\n9 chromedriver 0x000000010506930c chromedriver + 688908\n10 chromedriver 0x000000010506a88e chromedriver + 694414\n11 chromedriver 0x000000010543d1de chromedriver + 4702686\n12 chromedriver 0x0000000105441b19 chromedriver + 4721433\n13 chromedriver 0x000000010544928e chromedriver + 4752014\n14 chromedriver 0x000000010544291a chromedriver + 4725018\n15 chromedriver 0x0000000105416b02 chromedriver + 4545282\n16 chromedriver 0x0000000105461888 chromedriver + 4851848\n17 chromedriver 0x0000000105461a05 chromedriver + 4852229\n18 chromedriver 0x0000000105477e5f chromedriver + 4943455\n19 libsystem_pthread.dylib 0x00007ff8032fd4e1 _pthread_start + 125\n20 libsystem_pthread.dylib 0x00007ff8032f8f6b thread_start + 15\n",
|
||||
"output_type": "error",
|
||||
"traceback": [
|
||||
"\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
|
||||
"\u001b[0;31mTimeoutException\u001b[0m Traceback (most recent call last)",
|
||||
"Cell \u001b[0;32mIn[11], line 1\u001b[0m\n\u001b[0;32m----> 1\u001b[0m driver \u001b[39m=\u001b[39m scraper\u001b[39m.\u001b[39;49mload_topic(topic, headless\u001b[39m=\u001b[39;49m\u001b[39mTrue\u001b[39;49;00m)\n",
|
||||
"File \u001b[0;32m~/Projects/twitter_scraper/scraper.py:26\u001b[0m, in \u001b[0;36mload_topic\u001b[0;34m(topic, headless)\u001b[0m\n\u001b[1;32m 24\u001b[0m timeout \u001b[39m=\u001b[39m \u001b[39m30\u001b[39m \u001b[39m# seconds\u001b[39;00m\n\u001b[1;32m 25\u001b[0m xpath_str \u001b[39m=\u001b[39m \u001b[39m\"\u001b[39m\u001b[39m//*[text()=\u001b[39m\u001b[39m'\u001b[39m\u001b[39mNot now\u001b[39m\u001b[39m'\u001b[39m\u001b[39m]\u001b[39m\u001b[39m\"\u001b[39m\n\u001b[0;32m---> 26\u001b[0m element \u001b[39m=\u001b[39m WebDriverWait(driver, timeout)\u001b[39m.\u001b[39;49muntil(EC\u001b[39m.\u001b[39;49mpresence_of_element_located((By\u001b[39m.\u001b[39;49mXPATH, xpath_str)))\n\u001b[1;32m 27\u001b[0m element\u001b[39m.\u001b[39mclick()\n\u001b[1;32m 28\u001b[0m \u001b[39mreturn\u001b[39;00m driver\n",
|
||||
"File \u001b[0;32m~/Projects/twitter_scraper/venv/lib/python3.10/site-packages/selenium/webdriver/support/wait.py:95\u001b[0m, in \u001b[0;36mWebDriverWait.until\u001b[0;34m(self, method, message)\u001b[0m\n\u001b[1;32m 93\u001b[0m \u001b[39mif\u001b[39;00m time\u001b[39m.\u001b[39mmonotonic() \u001b[39m>\u001b[39m end_time:\n\u001b[1;32m 94\u001b[0m \u001b[39mbreak\u001b[39;00m\n\u001b[0;32m---> 95\u001b[0m \u001b[39mraise\u001b[39;00m TimeoutException(message, screen, stacktrace)\n",
|
||||
"\u001b[0;31mTimeoutException\u001b[0m: Message: \nStacktrace:\n0 chromedriver 0x000000010546ff38 chromedriver + 4910904\n1 chromedriver 0x00000001053efa03 chromedriver + 4385283\n2 chromedriver 0x0000000105034747 chromedriver + 472903\n3 chromedriver 0x000000010507934c chromedriver + 754508\n4 chromedriver 0x00000001050795a1 chromedriver + 755105\n5 chromedriver 0x00000001050bce94 chromedriver + 1031828\n6 chromedriver 0x000000010509f13d chromedriver + 909629\n7 chromedriver 0x00000001050ba28e chromedriver + 1020558\n8 chromedriver 0x000000010509eee3 chromedriver + 909027\n9 chromedriver 0x000000010506930c chromedriver + 688908\n10 chromedriver 0x000000010506a88e chromedriver + 694414\n11 chromedriver 0x000000010543d1de chromedriver + 4702686\n12 chromedriver 0x0000000105441b19 chromedriver + 4721433\n13 chromedriver 0x000000010544928e chromedriver + 4752014\n14 chromedriver 0x000000010544291a chromedriver + 4725018\n15 chromedriver 0x0000000105416b02 chromedriver + 4545282\n16 chromedriver 0x0000000105461888 chromedriver + 4851848\n17 chromedriver 0x0000000105461a05 chromedriver + 4852229\n18 chromedriver 0x0000000105477e5f chromedriver + 4943455\n19 libsystem_pthread.dylib 0x00007ff8032fd4e1 _pthread_start + 125\n20 libsystem_pthread.dylib 0x00007ff8032f8f6b thread_start + 15\n"
|
||||
]
|
||||
}
|
||||
],
|
||||
"source": [
|
||||
"driver = scraper.load_topic(topic, headless=True)"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
"metadata": {},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"# headless mode - has proven difficult...\n",
|
||||
"# database access\n",
|
||||
"# logging with warning to discord\n",
|
||||
"# main loop - go once every hour\n",
|
||||
"# containerize"
|
||||
]
|
||||
}
|
||||
],
|
||||
"metadata": {
|
||||
"kernelspec": {
|
||||
"display_name": "venv",
|
||||
"language": "python",
|
||||
"name": "python3"
|
||||
},
|
||||
"language_info": {
|
||||
"codemirror_mode": {
|
||||
"name": "ipython",
|
||||
"version": 3
|
||||
},
|
||||
"file_extension": ".py",
|
||||
"mimetype": "text/x-python",
|
||||
"name": "python",
|
||||
"nbconvert_exporter": "python",
|
||||
"pygments_lexer": "ipython3",
|
||||
"version": "3.10.6"
|
||||
},
|
||||
"orig_nbformat": 4,
|
||||
"vscode": {
|
||||
"interpreter": {
|
||||
"hash": "1d532f3642617da00cbdbdc5ef98bd8d4ca226c95ac593ade79d8cbc880a2afe"
|
||||
}
|
||||
}
|
||||
},
|
||||
"nbformat": 4,
|
||||
"nbformat_minor": 2
|
||||
}
|
||||
Reference in New Issue
Block a user