diff --git a/scraper.py b/scraper.py new file mode 100644 index 0000000..e77404b --- /dev/null +++ b/scraper.py @@ -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 diff --git a/test.ipynb b/test.ipynb new file mode 100644 index 0000000..e47b20c --- /dev/null +++ b/test.ipynb @@ -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 +}