→コード
>Fet-Fe (→コード: ツイートの画像を自動取得するため、勝手ながら改造させていただきました) |
>Fet-Fe (→コード) |
||
| 6行目: | 6行目: | ||
''' | ''' | ||
ver2.1. | ver2.1.1 2022/8/13恒心 | ||
当コードは恒心停止してしまったhttps://rentry.co/7298gの降臨ショーツイート自動収集スクリプトの復刻改善版です | 当コードは恒心停止してしまったhttps://rentry.co/7298gの降臨ショーツイート自動収集スクリプトの復刻改善版です | ||
| 23行目: | 23行目: | ||
ーーーーーーーーーーーーー【注意点】ーーーーーーーーーーーーー | ーーーーーーーーーーーーー【注意点】ーーーーーーーーーーーーー | ||
・Pythonのバージョンは3.10以上 | |||
・環境は玉葱前提です。 | ・環境は玉葱前提です。 | ||
・Whonix-Workstation, MacOSで動作確認済 | ・Whonix-Workstation, MacOSで動作確認済 | ||
| 42行目: | 43行目: | ||
import json | import json | ||
from time import sleep | from time import sleep | ||
from typing import Final | from typing import Final, NoReturn, TypeAlias | ||
from urllib.parse import quote, unquote | from urllib.parse import quote, unquote | ||
import warnings | import warnings | ||
| 52行目: | 53行目: | ||
##おそらくはツイートの内容によってMarkupResemblesLocatorWarningが吐き出されることがあるので無効化 | ##おそらくはツイートの内容によってMarkupResemblesLocatorWarningが吐き出されることがあるので無効化 | ||
warnings.simplefilter('ignore') | warnings.simplefilter('ignore') | ||
##型エイリアス | |||
Response: TypeAlias = requests.models.Response | |||
class TwitterArchiver: | class TwitterArchiver: | ||
| 82行目: | 86行目: | ||
##取得するツイート数の上限 | ##取得するツイート数の上限 | ||
LIMIT_N_TWEETS: Final[int] = | LIMIT_N_TWEETS: Final[int] = 100 | ||
##記録件数を報告するインターバル | ##記録件数を報告するインターバル | ||
| 133行目: | 137行目: | ||
#関数類 | #関数類 | ||
def __init__(self, krsw: bool=False): | def __init__(self, krsw: bool=False): | ||
self. | self._txt_data: str = '' | ||
self. | self._limit_count: int = 0 ##記録数 | ||
self._check_slash() ##スラッシュが抜けてないかチェック | self._check_slash() ##スラッシュが抜けてないかチェック | ||
self._check_tor() ##Ｔorが使えているかチェック | self._check_tor() ##Ｔorが使えているかチェック | ||
| 149行目: | 154行目: | ||
if krsw: | if krsw: | ||
print('クエリは自動的になしにナリます') | print('クエリは自動的になしにナリます') | ||
self._page: | self._page: Response | None = self._request(self.NITTER_INSTANCE + self.SEARCH_QUERY + name) | ||
if self._page is None: | if self._page is None: | ||
self._fail() | self._fail() | ||
else: | else: | ||
self._page: | self._page: Response = self._get_query(name) | ||
##終わりにするツイート取得 | ##終わりにするツイート取得 | ||
| 166行目: | 171行目: | ||
##坂根輝美に場所を知らせます | ##坂根輝美に場所を知らせます | ||
def _pickup_counter(self): | def _pickup_counter(self) -> None: | ||
print('ピックアップカウンター付近でふ') | print('ピックアップカウンター付近でふ') | ||
##引数のURLにHTTP接続します | ##引数のURLにHTTP接続します | ||
##失敗かどうかは呼出側で要判定 | ##失敗かどうかは呼出側で要判定 | ||
def _request_once(self, url: Final[str]) -> | def _request_once(self, url: Final[str]) -> Response: | ||
res = requests.get(url, timeout=self.REQUEST_TIMEOUT, headers=self.HEADERS, allow_redirects=False, proxies=self.PROXIES) | res: Response = requests.get(url, timeout=self.REQUEST_TIMEOUT, headers=self.HEADERS, allow_redirects=False, proxies=self.PROXIES) | ||
sleep(self.WAIT_TIME) ##DoS対策で待つ | sleep(self.WAIT_TIME) ##DoS対策で待つ | ||
return res | return res | ||
| 180行目: | 185行目: | ||
##接続失敗が何度も起きるとNoneを返します | ##接続失敗が何度も起きるとNoneを返します | ||
##呼出側で要None判定 | ##呼出側で要None判定 | ||
def _request(self, url: Final[str]) -> | def _request(self, url: Final[str]) -> Response | None: | ||
counter = 1 ##リクエスト挑戦回数を記録 | counter: int = 1 ##リクエスト挑戦回数を記録 | ||
while True: | while True: | ||
try: | try: | ||
res = self._request_once(url) ##リクエスト | res: Response = self._request_once(url) ##リクエスト | ||
res.raise_for_status() ##HTTPステータスコードが200番台以外でエラー発生 | res.raise_for_status() ##HTTPステータスコードが200番台以外でエラー発生 | ||
except requests.exceptions.RequestException as e: | except requests.exceptions.RequestException as e: | ||
| 198行目: | 203行目: | ||
##URLの最後にスラッシュ付いていなければ付ける | ##URLの最後にスラッシュ付いていなければ付ける | ||
##Twitterだけスラッシュ付いていないほうが都合いいので抜く | ##Twitterだけスラッシュ付いていないほうが都合いいので抜く | ||
def _check_slash(self) -> | def _check_slash(self) -> None | NoReturn: | ||
if self.NITTER_INSTANCE[-1] != '/': | if self.NITTER_INSTANCE[-1] != '/': | ||
raise RuntimeError('NITTER_INSTANCEの末尾には/が必須です') | raise RuntimeError('NITTER_INSTANCEの末尾には/が必須です') | ||
| 209行目: | 214行目: | ||
##Ｔorが使えているかチェック | ##Ｔorが使えているかチェック | ||
def _check_tor(self) -> | def _check_tor(self) -> None | NoReturn: | ||
print('Ｔorのチェック中ですを') | print('Ｔorのチェック中ですを') | ||
res = self._request_once('https://check.torproject.org/api/ip') ##リクエスト | res: Final[Response] = self._request_once('https://check.torproject.org/api/ip') ##リクエスト | ||
is_tor = json.loads(res.text)['IsTor'] | is_tor: Final[bool] = json.loads(res.text)['IsTor'] | ||
if is_tor: | if is_tor: | ||
print('Ｔor OK') | print('Ｔor OK') | ||
| 222行目: | 227行目: | ||
##接続を一回しか試さない_request_onceを使っているのは | ##接続を一回しか試さない_request_onceを使っているのは | ||
##激重インスタンスが指定されたとき試行回数増やして偶然成功してそのまま実行されるのを躱すため | ##激重インスタンスが指定されたとき試行回数増やして偶然成功してそのまま実行されるのを躱すため | ||
def _check_instance(self) -> | def _check_instance(self) -> None | NoReturn: | ||
print("nitterのインスタンスチェック中ですを") | print("nitterのインスタンスチェック中ですを") | ||
try: | try: | ||
res = self._request_once(self.NITTER_INSTANCE) ##リクエスト | res: Final[Response] = self._request_once(self.NITTER_INSTANCE) ##リクエスト | ||
res.raise_for_status() ##HTTPステータスコードが200番台以外でエラー発生 | res.raise_for_status() ##HTTPステータスコードが200番台以外でエラー発生 | ||
except requests.exceptions.RequestException as e: ##エラー発生時は終了 | except requests.exceptions.RequestException as e: ##エラー発生時は終了 | ||
| 233行目: | 238行目: | ||
##ツイート収集するユーザー名を取得 | ##ツイート収集するユーザー名を取得 | ||
##何も入力しないと尊師を指定するよう改良 | ##何も入力しないと尊師を指定するよう改良 | ||
def _get_name(self) -> | def _get_name(self) -> str | NoReturn: | ||
while True: | while True: | ||
print('アカウント名を入れなければない。空白だと自動的に' + self.CALLINSHOW + 'になりますを') | print('アカウント名を入れなければない。空白だと自動的に' + self.CALLINSHOW + 'になりますを') | ||
account_str = input() ##ユーザー入力受付 | account_str: str = input() ##ユーザー入力受付 | ||
##空欄で降臨ショー | ##空欄で降臨ショー | ||
if account_str == '': | if account_str == '': | ||
return self.CALLINSHOW | return self.CALLINSHOW | ||
else: | else: | ||
res = self._request(self.NITTER_INSTANCE + account_str) ##リクエストして結果取得 | res: Response | None = self._request(self.NITTER_INSTANCE + account_str) ##リクエストして結果取得 | ||
if res is None : ##リクエスト失敗判定 | if res is None : ##リクエスト失敗判定 | ||
self._fail() | self._fail() | ||
soup = BeautifulSoup(res.text, 'html.parser') ##beautifulsoupでレスポンス解析 | soup: BeautifulSoup = BeautifulSoup(res.text, 'html.parser') ##beautifulsoupでレスポンス解析 | ||
if soup.title == self.NITTER_ERROR_TITLE: ##タイトルがエラーでないか判定 | if soup.title == self.NITTER_ERROR_TITLE: ##タイトルがエラーでないか判定 | ||
print(account_str + "は実在の人物ではありませんでした") ##エラー時ループに戻る | print(account_str + "は実在の人物ではありませんでした") ##エラー時ループに戻る | ||
| 252行目: | 257行目: | ||
##検索クエリを取得 | ##検索クエリを取得 | ||
def _get_query(self, name: Final[str]) -> | def _get_query(self, name: Final[str]) -> Response | NoReturn: | ||
while True: | while True: | ||
print("検索クエリを入れるナリ。複数ある時は一つの塊ごとに入力するナリ。空欄を入れると検索開始ナリ。") | print("検索クエリを入れるナリ。複数ある時は一つの塊ごとに入力するナリ。空欄を入れると検索開始ナリ。") | ||
print("例：「無能」→改行→「脱糞」→改行→「弟殺し」→改行→空欄で改行") | print("例：「無能」→改行→「脱糞」→改行→「弟殺し」→改行→空欄で改行") | ||
query_str = [name] ##検索クエリ文字列をquery_strに収納。ユーザー名を先に含めておく | query_str: list[str] = [name] ##検索クエリ文字列をquery_strに収納。ユーザー名を先に含めておく | ||
query_input = input() ##ユーザー入力受付 | query_input: str = input() ##ユーザー入力受付 | ||
##空欄が押されるまでユーザー入力受付 | ##空欄が押されるまでユーザー入力受付 | ||
while query_input != '': | while query_input != '': | ||
query_str.append(quote(query_input)) | query_str.append(quote(query_input)) | ||
query_input = input() | query_input = input() | ||
res = self._request(self.NITTER_INSTANCE + self.SEARCH_QUERY + '+'.join(query_str)) ##リクエストして結果取得 | res: Response | None = self._request(self.NITTER_INSTANCE + self.SEARCH_QUERY + '+'.join(query_str)) ##リクエストして結果取得 | ||
if res is None : ##リクエスト失敗判定 | if res is None : ##リクエスト失敗判定 | ||
self._fail() | self._fail() | ||
soup = BeautifulSoup(res.text, 'html.parser') ##beautifulsoupでレスポンス解析 | soup: BeautifulSoup = BeautifulSoup(res.text, 'html.parser') ##beautifulsoupでレスポンス解析 | ||
##1件もないときはHTMLにtimeline-noneがあるから判定 | ##1件もないときはHTMLにtimeline-noneがあるから判定 | ||
if soup.find(class_='timeline-none') is None: | if soup.find(class_='timeline-none') is None: | ||
| 276行目: | 281行目: | ||
def _fail(self) -> NoReturn: | def _fail(self) -> NoReturn: | ||
print("接続失敗しすぎで強制終了ナリ") | print("接続失敗しすぎで強制終了ナリ") | ||
if self. | if self._txt_data != '': ##取得成功したデータがあれば発行 | ||
print("取得成功した分だけ発行しますを") | print("取得成功した分だけ発行しますを") | ||
self._make_txt() | self._make_txt() | ||
| 283行目: | 288行目: | ||
##テキスト発行 | ##テキスト発行 | ||
def _make_txt(self) -> NoReturn: | def _make_txt(self) -> NoReturn: | ||
result_txt: Final[str] = '{|class="wikitable" style="text-align: left;"\n' + self._txt_data + '|}' ##wikiの表の最初と最後 | |||
##ファイル出力 | ##ファイル出力 | ||
with codecs.open('tweet.txt', 'w', 'utf-8') as f: | with codecs.open('tweet.txt', 'w', 'utf-8') as f: | ||
f.write( | f.write(result_txt) | ||
print("テキストファイル手に入ったやで〜") | print("テキストファイル手に入ったやで〜") | ||
exit() ##終了 | exit() ##終了 | ||
| 293行目: | 298行目: | ||
def _stop_word(self) -> str: | def _stop_word(self) -> str: | ||
print(f"ここにツイートの本文を入れる！すると・・・・なんと一致するツイートで記録を中断する！（これは本当の仕様です。）ちなみに、空欄だと{self.LIMIT_N_TWEETS}件まで終了しない。") | print(f"ここにツイートの本文を入れる！すると・・・・なんと一致するツイートで記録を中断する！（これは本当の仕様です。）ちなみに、空欄だと{self.LIMIT_N_TWEETS}件まで終了しない。") | ||
end_str = input() ##ユーザー入力受付 | end_str: Final[str] = input() ##ユーザー入力受付 | ||
return end_str | return end_str | ||
##Twitterの画像をダウンロード | ##Twitterの画像をダウンロード | ||
def _download_media(self, media_name) -> bool: | def _download_media(self, media_name: Final[str]) -> bool: | ||
self. | os.makedirs(self.MEDIA_DIR, exist_ok=True) | ||
url = 'https://pbs.twimg.com/media/' + media_name | url: Final[str] = 'https://pbs.twimg.com/media/' + media_name | ||
res = self._request(url) | res: Final[Response | None] = self._request(url) | ||
if res is not None: | if res is not None: | ||
if 'image' not in res.headers['content-type']: | if 'image' not in res.headers['content-type']: | ||
| 317行目: | 316行目: | ||
def _get_tweet_media(self, tweet: bs4.element.Tag) -> str: | def _get_tweet_media(self, tweet: bs4.element.Tag) -> str: | ||
tweet_media = tweet.select_one('.tweet-body > .attachments') # 引用リツイート内のメディアを選択しないように.tweet-body直下の.attachmentsを選択 | tweet_media: bs4.element.Tag | None = tweet.select_one('.tweet-body > .attachments') # 引用リツイート内のメディアを選択しないように.tweet-body直下の.attachmentsを選択 | ||
media_txt = '' | media_txt: str = '' | ||
if tweet_media is not None: | if tweet_media is not None: | ||
media_list: list[str] = [] | |||
# ツイートの画像の取得 | |||
for image_a in tweet_media.select('.attachment.image a'): | |||
media_name: str = re.search(r'%2F([^%]*\.jpg)|%2F([^%]*\.jpeg)|%2F([^%]*\.png)|%2F([^%]*\.gif)', image_a.get('href')).group(1) | |||
media_list.append(f"[[ファイル:{media_name}|240px]]") | |||
if self._download_media(media_name): | |||
print(self.MEDIA_DIR + '/' + media_name + ' をアップロードしなければない。') | |||
else: | |||
print('https://pbs.twimg.com/media/' + media_name + ' をアップロードしなければない。') | |||
# 動画についてはm3u8で落ちてきて面倒臭いため取得しない | # 動画についてはm3u8で落ちてきて面倒臭いため取得しない | ||
""" | |||
for video in tweet_media.select('.attachment.video-container video'): | |||
media_url: str = unquote(re.search(r'[^\/]+$', video.get('data-url')).group(0)) | |||
print(media_url) | |||
""" | |||
media_txt = '<br>\n' + ' '.join(media_list) | |||
return media_txt | return media_txt | ||
def _get_tweet_quote(self, tweet: bs4.element.Tag) -> str: | def _get_tweet_quote(self, tweet: bs4.element.Tag) -> str: | ||
tweet_quote = tweet.select_one('.tweet-body > .quote.quote-big') # 引用リツイートを選択 | tweet_quote: Final[bs4.element.Tag | None] = tweet.select_one('.tweet-body > .quote.quote-big') # 引用リツイートを選択 | ||
quote_txt = '' | quote_txt: str = '' | ||
if tweet_quote is not None: | if tweet_quote is not None: | ||
link = tweet_quote.select_one('.quote-link').get('href') | link: str = tweet_quote.select_one('.quote-link').get('href') | ||
link = re.sub('#.*$', '', link) | link = re.sub('#.*$', '', link) | ||
link = self.TWITTER_URL + link | link = self.TWITTER_URL + link | ||
quote_txt = '<br>\n' + self._archive_url(link, link) | quote_txt: str = '<br>\n' + self._archive_url(link, link) | ||
return quote_txt | return quote_txt | ||
#ページからツイート本文をself. | #ページからツイート本文をself._txt_dataに収めていく | ||
def get_tweet(self) -> | def get_tweet(self) -> None | NoReturn: | ||
soup = BeautifulSoup(self._page.text, 'html.parser') ##beautifulsoupでレスポンス解析 | soup: Final[BeautifulSoup] = BeautifulSoup(self._page.text, 'html.parser') ##beautifulsoupでレスポンス解析 | ||
tweets = soup.find_all(class_='timeline-item') ##一ツイートのブロックごとにリストで取得 | tweets: Final[bs4.element.ResultSet] = soup.find_all(class_='timeline-item') ##一ツイートのブロックごとにリストで取得 | ||
for tweet in tweets: ##一ツイート毎に処理 | for tweet in tweets: ##一ツイート毎に処理 | ||
if tweet.a.text == self.NEWEST: ##Load Newestのボタンは処理しない | if tweet.a.text == self.NEWEST: ##Load Newestのボタンは処理しない | ||
| 356行目: | 356行目: | ||
if tweet.find(class_='retweet-header') is not None: ##retweet-headerはリツイートを示すので入っていれば処理しない | if tweet.find(class_='retweet-header') is not None: ##retweet-headerはリツイートを示すので入っていれば処理しない | ||
continue | continue | ||
tweet_url = self.TWITTER_URL + re.sub('#[^#]*$', '', tweet.find(class_='tweet-link').get('href')) ##ツイートのURL作成 | tweet_url: str = self.TWITTER_URL + re.sub('#[^#]*$', '', tweet.find(class_='tweet-link').get('href')) ##ツイートのURL作成 | ||
archived_tweet_url = self._callinshowlink_url(tweet_url) ##ツイートURLをテンプレートCallinShowlinkに変化 | archived_tweet_url: str = self._callinshowlink_url(tweet_url) ##ツイートURLをテンプレートCallinShowlinkに変化 | ||
tweet_content = tweet.find(class_='tweet-content media-body') ##ツイートの中身だけ取り出す | tweet_content: bs4.element.Tag = tweet.find(class_='tweet-content media-body') ##ツイートの中身だけ取り出す | ||
self._archive_soup(tweet_content) ##ツイートの中身のリンクをテンプレートArchiveに変化 | |||
media_txt = self._get_tweet_media(tweet) ##ツイートに画像などのメディアを追加 | media_txt: str = self._get_tweet_media(tweet) ##ツイートに画像などのメディアを追加 | ||
quote_txt = self._get_tweet_quote(tweet) ##引用リツイートの場合、元ツイートを追加 | quote_txt: str = self._get_tweet_quote(tweet) ##引用リツイートの場合、元ツイートを追加 | ||
self. | self._txt_data = '!' + archived_tweet_url + '\n|-\n|\n' + tweet_content.get_text().replace('\n', '<br>\n').replace('#', '<nowiki>#</nowiki>') + quote_txt + media_txt + '\n|-\n' + self._txt_data ##wikiの文法に変化 | ||
self. | self._limit_count += 1 ##記録回数をカウント | ||
if self. | if self._limit_count % self.REPORT_INTERVAL == 0: | ||
print(f"ツイートを{self. | print(f"ツイートを{self._limit_count}件も記録したンゴｗｗｗｗｗｗｗｗｗｗｗ") | ||
if self._stop != '' and self._stop in tweet_content.get_text(): ##目的ツイートか判定 | if self._stop != '' and self._stop in tweet_content.get_text(): ##目的ツイートか判定 | ||
print("目的ツイート発見でもう尾張屋根") | print("目的ツイート発見でもう尾張屋根") | ||
self._make_txt() | self._make_txt() | ||
if self. | if self._limit_count >= self.LIMIT_N_TWEETS: ##上限達成か判定 | ||
print(f"{self.LIMIT_N_TWEETS}件も記録している。もうやめにしませんか。") | print(f"{self.LIMIT_N_TWEETS}件も記録している。もうやめにしませんか。") | ||
self._make_txt() | self._make_txt() | ||
# | #tagをテンプレートArchiveの文字列に変化させる | ||
def _archive_soup(self, tag: bs4.element.Tag) -> | def _archive_soup(self, tag: bs4.element.Tag) -> None: | ||
urls_in_tweet: Final[bs4.element.ResultSet] = tag.find_all('a') | |||
for url in urls_in_tweet: | for url in urls_in_tweet: | ||
if re.match('^https?://', url.get('href')) is not None: ##先頭にhttpが付いていない物はハッシュタグの検索ページへのリンクなので処理しない | if re.match('^https?://', url.get('href')) is not None: ##先頭にhttpが付いていない物はハッシュタグの検索ページへのリンクなので処理しない | ||
if re.match('https' + self.NITTER_INSTANCE[4:], url.get('href')): | |||
url.replace_with( | #Nitter上のTwitterへのリンクを直す | ||
url_link = re.sub('https' + self.NITTER_INSTANCE[4:], self.TWITTER_URL + '/', url.get('href')) | |||
url_link = re.sub('\?.*$', '', url_link) | |||
url.replace_with(self._archive_url(url_link, url_link)) ##テンプレートArchiveに変化 | |||
elif re.match('piped.kavin.rocks/', url.text) or re.match('invidio.us/', url.text): | |||
#Nitter上のYouTubeへのリンクを直す | |||
url_link = url.get('href') | |||
url_link = re.sub('piped.kavin.rocks/', 'youtu.be/', url_link) | |||
url_link = re.sub('invidio.us/', 'youtu.be/', url_link) | |||
url.replace_with(self._archive_url(url_link, url_link)) ##テンプレートArchiveに変化 | |||
else: | |||
url.replace_with(self._archive_url(url.get('href'), url.get('href'))) ##テンプレートArchiveに変化 | |||
#URLをテンプレートArchiveに変化させる | #URLをテンプレートArchiveに変化させる | ||
| 393行目: | 402行目: | ||
##URLから魚拓返す | ##URLから魚拓返す | ||
def _archive(self, url: Final[str]) -> str: | def _archive(self, url: Final[str]) -> str: | ||
archive_url = self.ARCHIVE_TODAY_STANDARD + url ##wikiに載せるとき用URLで失敗するとこのままhttps://archive.ph/https://xxxxxxxxの形で返される | archive_url: str = self.ARCHIVE_TODAY_STANDARD + re.sub('#', '%23', url) ##wikiに載せるとき用URLで失敗するとこのままhttps://archive.ph/https://xxxxxxxxの形で返される | ||
res = self._request(self.ARCHIVE_TODAY + url) ##アクセス用URL使って結果を取得 | res: Final[Response | None] = self._request(self.ARCHIVE_TODAY + re.sub('#', '%23', url)) ##アクセス用URL使って結果を取得 | ||
if res is None : ##魚拓接続失敗時処理 | if res is None : ##魚拓接続失敗時処理 | ||
print(archive_url + 'にアクセス失敗ナリ。出力されるテキストにはそのまま記載されるナリ。') | print(archive_url + 'にアクセス失敗ナリ。出力されるテキストにはそのまま記載されるナリ。') | ||
else: | else: | ||
soup = BeautifulSoup(res.text, 'html.parser') ##beautifulsoupでレスポンス解析 | soup: Final[BeautifulSoup] = BeautifulSoup(res.text, 'html.parser') ##beautifulsoupでレスポンス解析 | ||
content = soup.find(id="CONTENT") ##archive.todayの魚拓一覧ページの中身だけ取得 | content: bs4.element.Tag = soup.find(id="CONTENT") ##archive.todayの魚拓一覧ページの中身だけ取得 | ||
if content.get_text()[:len(self.NO_ARCHIVE)] == self.NO_ARCHIVE: ##魚拓があるかないか判定 | if content.get_text()[:len(self.NO_ARCHIVE)] == self.NO_ARCHIVE: ##魚拓があるかないか判定 | ||
print(url + "の魚拓がない。これはいけない。") | print(url + "の魚拓がない。これはいけない。") | ||
| 407行目: | 416行目: | ||
##新しいページを取得 | ##新しいページを取得 | ||
def go_to_new_page(self) -> | def go_to_new_page(self) -> None | NoReturn: | ||
soup = BeautifulSoup(self._page.text, 'html.parser') ##beautifulsoupでレスポンス解析 | soup: Final[BeautifulSoup] = BeautifulSoup(self._page.text, 'html.parser') ##beautifulsoupでレスポンス解析 | ||
show_mores: Final[bs4.element.ResultSet] = soup.find_all(class_="show-more") | |||
for | for show_more in show_mores: ##show-moreに次ページへのリンクか前ページへのリンクがある | ||
if | if show_more.text != self.NEWEST: ##前ページへのリンクではないか判定 | ||
new_url: str = self.NITTER_INSTANCE + self.SEARCH + show_more.a.get('href') ##直下のaタグのhrefの中身取ってURL頭部分と合体 | |||
res = self._request( | res: Final[Response | None] = self._request(new_url) ##接続してHTML取ってくる | ||
if res is None: | if res is None: | ||
self._fail() | self._fail() | ||
new_page_soup: Final[BeautifulSoup] = BeautifulSoup(res.text, 'html.parser') ##beautifulsoupでレスポンス解析 | |||
if | if new_page_soup.find(class_="timeline-end") is None: ##ツイートの終端ではtimeline-endだけのページになるので判定 | ||
print(res.url + 'に移動しますを') | print(res.url + 'に移動しますを') | ||
self._page = res ##まだ残りツイートがあるのでページを返して再度ツイート本文収集 | self._page = res ##まだ残りツイートがあるのでページを返して再度ツイート本文収集 | ||
| 426行目: | 435行目: | ||
if __name__ == '__main__': | if __name__ == '__main__': | ||
krsw = len(sys.argv) > 1 and sys.argv[1] == 'krsw' ##コマンドライン引数があるかどうかのフラグ | krsw: Final[bool] = len(sys.argv) > 1 and sys.argv[1] == 'krsw' ##コマンドライン引数があるかどうかのフラグ | ||
twitter_archiver = TwitterArchiver(krsw) | twitter_archiver: TwitterArchiver = TwitterArchiver(krsw) | ||
##ツイートを取得し終えるまでループ | ##ツイートを取得し終えるまでループ | ||
while True: | while True: | ||
twitter_archiver.get_tweet() ##self. | twitter_archiver.get_tweet() ##self._txt_dataにページ内のリツイート以外の全ツイートの中身突っ込んでいく | ||
twitter_archiver.go_to_new_page() ##新しいページ取得 | twitter_archiver.go_to_new_page() ##新しいページ取得 | ||
</syntaxhighlight> | </syntaxhighlight> | ||