from selenium import webdriver from selenium.webdriver.common.action_chains import ActionChains from selenium.webdriver.chrome.service import Service from webdriver_manager.chrome import ChromeDriverManager from selenium.webdriver.common.by import By from selenium.webdriver.support.wait import WebDriverWait import selenium.webdriver.support.expected_conditions as EC from selenium.webdriver.remote.webelement import WebElement from pydantic import BaseModel, validator, parse_obj_as, AnyHttpUrl from typing import Union, List from datetime import datetime from PIL import Image from io import BytesIO import requests import pyperclip import logging import time import traceback class Twitter(object): def __init__(self): self.webdriver = None self.actiondriver = None self.timeline_elem = None self._read_posts_timestamp_list = list() self._unread_posts_element_list = list() @staticmethod def _url( account: str, date_begin: datetime, date_end: datetime) -> AnyHttpUrl: return parse_obj_as(AnyHttpUrl, f"https://twitter.com/search?q=(from%3A{account})%20until%3A{date_end.strftime('%Y-%m-%d')}%20since%3A{date_begin.strftime('%Y-%m-%d')}%20-filter%3Areplies&src=typed_query&f=top") def setup(self) -> None: # setup webdriver driver = webdriver.Chrome(service=Service(ChromeDriverManager().install())) self.webdriver = driver self.actiondriver = ActionChains(self.webdriver) def close(self) -> None: self.webdriver.close() def __enter__(self): self.setup() return self def __exit__(self, exc_type, exc_value, traceback): self.close() def _update_unread_posts(self) -> None: self._handle_popup() elem_list = self.timeline_elem.find_elements(By.XPATH, """.//article[@data-testid="tweet"]""") # elem_list = self.webdriver.find_elements(By.CSS_SELECTOR, '[data-testid="tweet"]') for elem in elem_list: try: elem_list = elem.find_elements(By.XPATH, './/time') assert len(elem_list) > 0, 'failed locating time of post' timestamp_str = elem_list[0].get_attribute('datetime') except Exception: pass else: # print('considering post with timestamp_str: ', timestamp_str) if timestamp_str not in self._read_posts_timestamp_list: self._unread_posts_element_list.append(elem) # print('attached post to list') else: # print('post already in list') pass def unread_posts_available(self) -> bool: if len(self._unread_posts_element_list) == 0: self._update_unread_posts() return len(self._unread_posts_element_list) > 0 def number_of_posts_read(self) -> int: return len(self._read_posts_timestamp_list) def _handle_popup(self): elem_list = self.webdriver.find_elements(By.XPATH, "//*[text()='Not now']") if len(elem_list) > 0: elem_list[0].click() def _accept_cookies(self): self._handle_popup() try: elem = self.webdriver.find_element(By.XPATH, "//*[text()='Accept all cookies']") elem.click() except Exception: logging.warning('failed accepting cookies') def _open_page(self, account: str, date_begin: datetime, date_end: datetime) -> None: url = self._url(account,date_begin, date_end) self.webdriver.get(str(url)) # wait for popup to appear and click on 'not now'-button try: elem = WebDriverWait( driver=self.webdriver, timeout=60 ).until( EC.presence_of_element_located( (By.XPATH, "//*[text()='Not now']") ) ) except TimeoutError: pass else: elem.click() # click allow cookies self._accept_cookies() # identify timeline element self.timeline_elem = self.webdriver.find_element(By.XPATH, """//div[@aria-label="Timeline: Search timeline"]""") # update unread posts self._update_unread_posts() def _extract_account(self, elem): self._handle_popup() try: elem_list = elem.find_elements(By.XPATH, ".//span[contains(text(), '@')]") assert len(elem_list) > 0, 'failed locating account' self.actiondriver.move_to_element(elem_list[0]).perform() account_str = elem_list[0].text except Exception: logging.error('failed extracting account name') account_str = '' return account_str def _extract_time(self, elem): self._handle_popup() try: elem_list = elem.find_elements(By.XPATH, './/time') assert len(elem_list) > 0, 'failed locating time of post' self.actiondriver.move_to_element(elem_list[0]).perform() datetime_str = elem_list[0].get_attribute('datetime') except Exception: logging.error('failed extracting post time') datetime_str = '' return datetime_str def _extract_text(self, elem): self._handle_popup() try: elem_list = elem.find_elements(By.XPATH, './/div[@data-testid="tweetText"]') assert len(elem_list) > 0, 'failed locating text of post' self.actiondriver.move_to_element(elem_list[0]).perform() text = elem_list[0].text except Exception: logging.error('failed extracting post text') text = '' return text def _extract_post_url(self, elem): self._handle_popup() try: share_button = elem.find_element(By.XPATH, """.//div[@aria-label="Share Tweet"]""") # WebDriverWait(self.webdriver, 15).until(EC.element_to_be_clickable((By.XPATH, "//div[@aria-label='Share Tweet']"))).click() # self.webdriver.execute_script("arguments[0].scrollIntoView();", elem_list[0]) # scroll to element self.webdriver.execute_script( """arguments[0].scrollIntoView({behavior: "smooth", block: "center", inline: "nearest"})""", share_button ) # move mouse to element self.actiondriver.move_to_element(share_button).perform() # click element share_button = elem.find_element(By.XPATH, """.//div[@aria-label="Share Tweet"]""") share_button.click() # wait for popup button to appear share_button = WebDriverWait( driver=self.webdriver, timeout=10 ).until( EC.presence_of_element_located( (By.XPATH, """//span[contains(text(), "Copy link to Tweet")]""") ) ) # click button # elem_list = elem.find_elements(By.XPATH, "//span[contains(text(), 'Copy link to Tweet')]") self.actiondriver.move_to_element(share_button).perform() share_button.click() # paste copied url url = pyperclip.paste() except Exception: logging.error('failed extracting post url') url = '' return url def _extract_video_url(self, elem): self._handle_popup() try: elem_list = elem.find_elements(By.TAG_NAME, 'video') if len(elem_list) > 0: video_url = elem_list[0].get_attribute('src') else: video_url = '' except Exception: logging.error('failed extracting post video') video_url = '' return video_url def _extract_post_images(self, elem): self._handle_popup() img_list = list() try: elem_list = elem.find_elements(By.TAG_NAME, 'img') if len(elem_list) == 0: return img_list # get image urls image_url_list = [e.get_attribute('src') for e in elem_list] # keep only media-related urls image_url_list = [url for url in image_url_list if 'media' in url] if len(image_url_list) == 0: return img_list # download image data for url in image_url_list: try: data = requests.get(url).content except Exception: logging.error(f'failed downloading image: {url}') continue img_data = BytesIO(data) img_list.append(img_data) except Exception: logging.error('failed extracting post images') return img_list def get_post(self): if len(self._unread_posts_element_list) == 0: raise ValueError('no unread posts to read') # get element elem = self._unread_posts_element_list.pop(0) # prepare response content = dict() try: # look for popup self._handle_popup() # scroll to element self.actiondriver.move_to_element(elem).perform() # get account content['account'] = self._extract_account(elem) # get time datetime_str = self._extract_time(elem) # print(datetime_str) content['time'] = datetime.strptime(datetime_str, '%Y-%m-%dT%H:%M:%S.%fZ') # get text content['text'] = self._extract_text(elem) # get post link content['url'] = self._extract_post_url(elem) # get video url content['video'] = self._extract_video_url(elem) # get image(s) content['images'] = self._extract_post_images(elem) except Exception: traceback.print_exc() else: # add timestamp_str to list of read posts self._read_posts_timestamp_list.append(datetime_str) # print('sucessfully read post') return content, elem # def extract_tweet(webelement: WebElement) -> dict: # post_time = webelement.find_element(By.XPATH, '//time').get_attribute('datetime') # post_text = webelement.find_element(By.CSS_SELECTOR, '[data-testid="tweetText"]').text # # find images in webelement # img_url_list = [elem.get_attribute('src') for elem in webelement.find_elements(By.TAG_NAME, 'img')] # # keep all images with url containing 'media' # img_url_list = [url for url in img_url_list if 'media' in url] # img_data_list = [requests.get(img_url).content for img_url in img_url_list] # tweet = Tweet( # time=post_time, # text=post_text, # images=img_data_list # ) # return tweet # def fetch_period(account: str, date_begin: datetime, date_end: datetime) -> List[Tweet]: # # prepare url # url = base_url(account, date_begin, date_end) # # setup webdriver # driver = webdriver.Chrome('../chromedriver_mac') # driver.get(str(url)) # # wait for popup to appear # elem = WebDriverWait( # driver=driver, # timeout=20 # ).until( # EC.presence_of_element_located( # (By.XPATH, "//*[text()='Not now']") # ) # ) # # turn off notifications - click not now button # elem.click() # # click allow cookies # elem = driver.find_element(By.XPATH, "//*[text()='Accept all cookies']") # elem.click() #