297 lines
11 KiB
Python
297 lines
11 KiB
Python
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()
|
|
#
|