マヨケーがポアされたため、現在はロシケーがメインとなっています。

「利用者:夜泣き/スクリプト」の版間の差分

提供:唐澤貴洋Wiki
ナビゲーションに移動 検索に移動
>Fet-Fe
(→‎コード: ツイートの画像を自動取得するため、勝手ながら改造させていただきました)
>Fet-Fe
6行目: 6行目:


'''
'''
ver2.1.0 2022/8/8恒心
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, Optional, Union, NoReturn, ClassVar
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] = 300
   LIMIT_N_TWEETS: Final[int] = 100


   ##記録件数を報告するインターバル
   ##記録件数を報告するインターバル
133行目: 137行目:
   #関数類
   #関数類
   def __init__(self, krsw: bool=False):
   def __init__(self, krsw: bool=False):
     self.txt_data = ''
     self._txt_data: str = ''
     self.limit_count = 0 ##記録数
     self._limit_count: int = 0 ##記録数
 
     self._check_slash() ##スラッシュが抜けてないかチェック
     self._check_slash() ##スラッシュが抜けてないかチェック
     self._check_tor() ##Torが使えているかチェック
     self._check_tor() ##Torが使えているかチェック
149行目: 154行目:
     if krsw:
     if krsw:
       print('クエリは自動的になしにナリます')
       print('クエリは自動的になしにナリます')
       self._page: str = self._request(self.NITTER_INSTANCE + self.SEARCH_QUERY + name)
       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: str = self._get_query(name)
       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]) -> requests.models.Response:
   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]) -> Optional[requests.models.Response]:
   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) -> Union[None, NoReturn]:
   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行目:


   ##Torが使えているかチェック
   ##Torが使えているかチェック
   def _check_tor(self) -> Union[None, NoReturn]:
   def _check_tor(self) -> None | NoReturn:
     print('Torのチェック中ですを')
     print('Torのチェック中ですを')
     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('Tor OK')
       print('Tor OK')
222行目: 227行目:
   ##接続を一回しか試さない_request_onceを使っているのは
   ##接続を一回しか試さない_request_onceを使っているのは
   ##激重インスタンスが指定されたとき試行回数増やして偶然成功してそのまま実行されるのを躱すため
   ##激重インスタンスが指定されたとき試行回数増やして偶然成功してそのまま実行されるのを躱すため
   def _check_instance(self) -> Union[None, NoReturn]:
   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) -> Union[str, NoReturn]:
   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]) -> Union[requests.models.Response, NoReturn]:
   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.txt_data != '': ##取得成功したデータがあれば発行
     if self._txt_data != '': ##取得成功したデータがあれば発行
       print("取得成功した分だけ発行しますを")
       print("取得成功した分だけ発行しますを")
       self._make_txt()
       self._make_txt()
283行目: 288行目:
   ##テキスト発行
   ##テキスト発行
   def _make_txt(self) -> NoReturn:
   def _make_txt(self) -> NoReturn:
     self.txt_data = '{|class="wikitable" style="text-align: left;"\n' + self.txt_data + '|}' ##wikiの表の最初と最後
     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(self.txt_data)
       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
  def _mkdir(self):
    try:
      os.mkdir(self.MEDIA_DIR)
    except FileExistsError:
      pass


   ##Twitterの画像をダウンロード
   ##Twitterの画像をダウンロード
   def _download_media(self, media_name) -> bool:
   def _download_media(self, media_name: Final[str]) -> bool:
     self._mkdir()
     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:
       for tweet_medium in tweet_media:
       media_list: list[str] = []
        # ツイートの画像の取得
      # ツイートの画像の取得
        if tweet_medium.select_one('.attachment.image a'):
      for image_a in tweet_media.select('.attachment.image a'):
          media_name = re.search(r'%2F([^%]*\.jpg)|%2F([^%]*\.jpeg)|%2F([^%]*\.png)|%2F([^%]*\.gif)', tweet_medium.select_one('.attachment.image a').get('href')).group(1)
        media_name: str = re.search(r'%2F([^%]*\.jpg)|%2F([^%]*\.jpeg)|%2F([^%]*\.png)|%2F([^%]*\.gif)', image_a.get('href')).group(1)
          media_txt += f"<br>\n[[ファイル:{media_name}|240px]]"
        media_list.append(f"[[ファイル:{media_name}|240px]]")
          if self._download_media(media_name):
        if self._download_media(media_name):
            print(self.MEDIA_DIR + '/' + media_name + ' をアップロードしなければない。')
          print(self.MEDIA_DIR + '/' + media_name + ' をアップロードしなければない。')
          else:
        else:
            print('https://pbs.twimg.com/media/' + media_name + ' をアップロードしなければない。')
          print('https://pbs.twimg.com/media/' + media_name + ' をアップロードしなければない。')
         # 動画についてはm3u8で落ちてきて面倒臭いため取得しない
         # 動画についてはm3u8で落ちてきて面倒臭いため取得しない
        """
      """
        elif tweet_medium.select_one('.attachment.video-container video'):
      for video in tweet_media.select('.attachment.video-container video'):
          media_url = unquote(re.search(r'[^\/]+$', tweet_medium.select_one('.attachment.video-container video').get('data-url')).group(0))
        media_url: str = unquote(re.search(r'[^\/]+$', video.get('data-url')).group(0))
          print(media_url)
        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.txt_dataに収めていく
   #ページからツイート本文をself._txt_dataに収めていく
   def get_tweet(self) -> Union[None, NoReturn]:
   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') ##ツイートの中身だけ取り出す
       tweet_content = self._archive_soup(tweet_content) ##ツイートの中身のリンクをテンプレートArchiveに変化
       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.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._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.limit_count += 1 ##記録回数をカウント
       self._limit_count += 1 ##記録回数をカウント
       if self.limit_count % self.REPORT_INTERVAL == 0:
       if self._limit_count % self.REPORT_INTERVAL == 0:
         print(f"ツイートを{self.limit_count}件も記録したンゴwwwwwwwwwww")
         print(f"ツイートを{self._limit_count}件も記録したンゴwwwwwwwwwww")
       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.limit_count >= self.LIMIT_N_TWEETS: ##上限達成か判定
       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()


   #soupをテンプレートArchiveに変化させる
   #tagをテンプレートArchiveの文字列に変化させる
   def _archive_soup(self, tag: bs4.element.Tag) -> BeautifulSoup:
   def _archive_soup(self, tag: bs4.element.Tag) -> None:
     soup = BeautifulSoup(str(tag))
     urls_in_tweet: Final[bs4.element.ResultSet] = tag.find_all('a')
    urls_in_tweet = soup.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が付いていない物はハッシュタグの検索ページへのリンクなので処理しない
         newstr = soup.new_string(self._archive_url(url.get('href'), url.text)) ##テンプレートArchiveの文字列作成
         if re.match('https' + self.NITTER_INSTANCE[4:], url.get('href')):
         url.replace_with(newstr) ##テンプレートArchiveに変化
          #Nitter上のTwitterへのリンクを直す
    return soup
          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) -> Union[None, NoReturn]:
   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でレスポンス解析
     showmores = soup.find_all(class_="show-more")
     show_mores: Final[bs4.element.ResultSet] = soup.find_all(class_="show-more")
     for showmore in showmores: ##show-moreに次ページへのリンクか前ページへのリンクがある
     for show_more in show_mores: ##show-moreに次ページへのリンクか前ページへのリンクがある
       if showmore.text != self.NEWEST:  ##前ページへのリンクではないか判定
       if show_more.text != self.NEWEST:  ##前ページへのリンクではないか判定
         newurl = self.NITTER_INSTANCE + self.SEARCH + showmore.a.get('href') ##直下のaタグのhrefの中身取ってURL頭部分と合体
         new_url: str = self.NITTER_INSTANCE + self.SEARCH + show_more.a.get('href') ##直下のaタグのhrefの中身取ってURL頭部分と合体
     res = self._request(newurl) ##接続してHTML取ってくる
     res: Final[Response | None] = self._request(new_url) ##接続してHTML取ってくる
     if res is None:
     if res is None:
       self._fail()
       self._fail()
     newsoup = BeautifulSoup(res.text, 'html.parser') ##beautifulsoupでレスポンス解析
     new_page_soup: Final[BeautifulSoup] = BeautifulSoup(res.text, 'html.parser') ##beautifulsoupでレスポンス解析
     if newsoup.find(class_="timeline-end") is None: ##ツイートの終端ではtimeline-endだけのページになるので判定
     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.txt_dataにページ内のリツイート以外の全ツイートの中身突っ込んでいく
     twitter_archiver.get_tweet() ##self._txt_dataにページ内のリツイート以外の全ツイートの中身突っ込んでいく
     twitter_archiver.go_to_new_page() ##新しいページ取得
     twitter_archiver.go_to_new_page() ##新しいページ取得
‎</syntaxhighlight>
‎</syntaxhighlight>

2022年8月13日 (土) 23:02時点における版

‎とりあえず取り急ぎ。バグ報告は利用者・トーク:夜泣き

コード

#!/usr/bin/env python3

'''
ver2.1.1 2022/8/13恒心

当コードは恒心停止してしまったhttps://rentry.co/7298gの降臨ショーツイート自動収集スクリプトの復刻改善版です
前開発者との出会いに感謝

〜〜〜〜〜〜〜〜〜〜〜〜〜【使い方】〜〜〜〜〜〜〜〜〜〜〜〜〜

・terminalに $ python3 (ファイル名) で作動します
・定数類は状況に応じて変えてください
・$ python3 (ファイル名) krsw コマンドライン引数をkrswとつけると自動モードです
・自動モードではユーザーは降臨ショー、クエリはなし、取得上限まで自動です
・つまりユーザー入力が要りません

〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜

ーーーーーーーーーーーーー【注意点】ーーーーーーーーーーーーー

・Pythonのバージョンは3.10以上
・環境は玉葱前提です。
・Whonix-Workstation, MacOSで動作確認済
  ・MacOSの場合はbrewでtorコマンドを導入し、実行
・PySocks, bs4はインストールしないと標準で入ってません
・requestsも環境によっては入っていないかもしれない
・$ pip install bs4 requests PySocks
・pipも入っていなければ$ sudo apt install pip
・バグ報告はhttps://krsw-wiki.org/wiki/?curid=15799#利用者:夜泣き/スクリプトについて

ーーーーーーーーーーーーーーーーーーーーーーーーーーーーーーー
'''

#インポート類
import sys
import os
import codecs
import re
import json
from time import sleep
from typing import Final, NoReturn, TypeAlias
from urllib.parse import quote, unquote
import warnings

import requests
import bs4
from bs4 import BeautifulSoup

##おそらくはツイートの内容によってMarkupResemblesLocatorWarningが吐き出されることがあるので無効化
warnings.simplefilter('ignore')

##型エイリアス
Response: TypeAlias = requests.models.Response

class TwitterArchiver:
  #定数・設定類

  ##nitterのインスタンス
  ##生きているのはhttps://github.com/zedeus/nitter/wiki/Instancesで確認
  ##末尾にスラッシュ必須
  NITTER_INSTANCE: Final[str] = 'http://ibsboeui2im5o7dxnik3s5yghufumgy5abevtij5nbizequfpu4qi4ad.onion/'

  ##archive.todayの魚拓
  ##実際にアクセスして魚拓があるか調べるのにはonionを使用
  ##末尾にスラッシュ必須
  ARCHIVE_TODAY: Final[str] = 'http://archiveiya74codqgiixo33q62qlrqtkgmcitqx5u2oeqnmn5bpcbiyd.onion/'

  ##archive.todayの魚拓
  ##記事の文章に使うのはクリアネット
  ##末尾にスラッシュ必須
  ARCHIVE_TODAY_STANDARD: Final[str] = 'https://archive.ph/'

  ##twitterのURL
  ##末尾にスラッシュ禁止
  TWITTER_URL: Final[str] = 'https://twitter.com'

  ##降臨ショーのユーザーネーム
  CALLINSHOW: Final[str] = 'CallinShow'

  ##HTTPリクエストのタイムアウト秒数
  REQUEST_TIMEOUT: Final[int] = 30

  ##取得するツイート数の上限
  LIMIT_N_TWEETS: Final[int] = 100

  ##記録件数を報告するインターバル
  REPORT_INTERVAL: Final[int] = 5

  ##HTTPリクエスト失敗時の再試行回数
  LIMIT_N_REQUESTS: Final[int] = 5

  ##HTTPリクエスト失敗時にさらに追加する待機時間
  WAIT_TIME_FOR_ERROR: Final[int] = 4

  ##HTTPリクエスト成功失敗関わらず待機時間
  ##1秒待つだけで行儀がいいクローラーだそうなので既定では1秒
  ##しかし日本のポリホーモは1秒待っていても捕まえてくるので注意
  ##https://ja.wikipedia.org/wiki/?curid=2187212
  WAIT_TIME: Final[int] = 1

  ##nitterのURLのドメインとユーザーネーム部分の接続部品
  SEARCH_QUERY: Final[str] = 'search?q=from%3A'

  ##HTTPリクエスト時のユーザーエージェントとヘッダ
  HEADERS: Final[dict[str, str]] = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; rv:91.0) Gecko/20100101 Firefox/91.0'
  }

  ##HTTP
  PROXIES: Final[dict[str, str]] = {
    'http': 'socks5h://127.0.0.1:9050',
    'https': 'socks5h://127.0.0.1:9050'
  }

  ##nitterでユーザーがいなかったとき返ってくるページのタイトル
  ##万が一仕様変更で変わったとき用
  NITTER_ERROR_TITLE: Final[str] = 'Error|nitter'

  ##archive.todayで魚拓がなかったときのレスポンス
  ##万が一仕様変更で変わったとき用
  NO_ARCHIVE: Final[str] = 'No results'

  ##nitterで次ページ遷移時のURLから抜け落ちてる部分
  SEARCH: Final[str] = 'search'

  ##nitterの前ページ読み込み部分の名前
  ##万が一仕様変更で変わったとき用
  NEWEST: Final[str] = 'Load newest'

  ##画像などのツイートメディアをダウンロードするディレクトリ
  MEDIA_DIR: Final[str] = 'tweet_media'

  #関数類
  def __init__(self, krsw: bool=False):
    self._txt_data: str = ''
    self._limit_count: int = 0 ##記録数

    self._check_slash() ##スラッシュが抜けてないかチェック
    self._check_tor() ##Torが使えているかチェック
    self._check_instance() ##インスタンスが死んでないかチェック

    ##ユーザー名取得
    if krsw:
      print('名前は自動的に' + self.CALLINSHOW + 'にナリます')
      name: Final[str] = self.CALLINSHOW
    else:
      name: Final[str] = self._get_name()

    ##検索クエリとページ取得
    if krsw:
      print('クエリは自動的になしにナリます')
      self._page: Response | None = self._request(self.NITTER_INSTANCE + self.SEARCH_QUERY + name)
      if self._page is None:
        self._fail()
    else:
      self._page: Response = self._get_query(name)

    ##終わりにするツイート取得
    if krsw:
      print('終わりにするツイートは自動的になしにナリます')
      self._stop: Final[str] = ''
    else:
      self._stop: Final[str] = self._stop_word()

    print()


  ##坂根輝美に場所を知らせます
  def _pickup_counter(self) -> None:
    print('ピックアップカウンター付近でふ')

  ##引数のURLにHTTP接続します
  ##失敗かどうかは呼出側で要判定
  def _request_once(self, url: Final[str]) -> Response:
    res: Response = requests.get(url, timeout=self.REQUEST_TIMEOUT, headers=self.HEADERS, allow_redirects=False, proxies=self.PROXIES)
    sleep(self.WAIT_TIME) ##DoS対策で待つ
    return res

  ##HTTP接続を再試行回数まで試します
  ##成功すると結果を返します
  ##接続失敗が何度も起きるとNoneを返します
  ##呼出側で要None判定
  def _request(self, url: Final[str]) -> Response | None:
    counter: int = 1 ##リクエスト挑戦回数を記録
    while True:
      try:
        res: Response = self._request_once(url) ##リクエスト
        res.raise_for_status() ##HTTPステータスコードが200番台以外でエラー発生
      except requests.exceptions.RequestException as e:
        print(url + 'への通信失敗ナリ  ' + f"{counter}/{self.LIMIT_N_REQUESTS}回")
        if counter < self.LIMIT_N_REQUESTS: ##エラー発生時上限まで再挑戦
          counter += 1 ##現在の試行回数1回増やす
          sleep(self.WAIT_TIME_FOR_ERROR) ##失敗時は長めに待つ
        else:
          return None ##失敗したらNone返却し呼出側で対処してもらう
      else:
        return res ##リクエストの結果返す

  ##URLの最後にスラッシュ付いていなければ付ける
  ##Twitterだけスラッシュ付いていないほうが都合いいので抜く
  def _check_slash(self) -> None | NoReturn:
    if self.NITTER_INSTANCE[-1] != '/':
      raise RuntimeError('NITTER_INSTANCEの末尾には/が必須です')
    if self.ARCHIVE_TODAY[-1] != '/':
      raise RuntimeError('ARCHIVE_TODAYの末尾には/が必須です')
    if self.ARCHIVE_TODAY_STANDARD[-1] != '/':
      raise RuntimeError('ARCHIVE_TODAY_STANDARDの末尾には/が必須です')
    if self.TWITTER_URL[-1] == '/':
      raise RuntimeError('TWITTER_URLの末尾には/があってはいけません')

  ##Torが使えているかチェック
  def _check_tor(self) -> None | NoReturn:
    print('Torのチェック中ですを')
    res: Final[Response] = self._request_once('https://check.torproject.org/api/ip') ##リクエスト
    is_tor: Final[bool] = json.loads(res.text)['IsTor']
    if is_tor:
      print('Tor OK')
    else:
      raise RuntimeError('通信がTorを経由していないなりを')

  ##nitterのインスタンスが生きているかチェック
  ##死んでいたらそこで終了
  ##接続を一回しか試さない_request_onceを使っているのは
  ##激重インスタンスが指定されたとき試行回数増やして偶然成功してそのまま実行されるのを躱すため
  def _check_instance(self) -> None | NoReturn:
    print("nitterのインスタンスチェック中ですを")
    try:
      res: Final[Response] = self._request_once(self.NITTER_INSTANCE) ##リクエスト
      res.raise_for_status() ##HTTPステータスコードが200番台以外でエラー発生
    except requests.exceptions.RequestException as e: ##エラー発生時は終了
      print('インスタンスが死んでますを')
      exit()

  ##ツイート収集するユーザー名を取得
  ##何も入力しないと尊師を指定するよう改良
  def _get_name(self) -> str | NoReturn:
    while True:
      print('アカウント名を入れなければない。空白だと自動的に' + self.CALLINSHOW + 'になりますを')
      account_str: str = input() ##ユーザー入力受付
      ##空欄で降臨ショー
      if account_str == '':
        return self.CALLINSHOW
      else:
        res: Response | None = self._request(self.NITTER_INSTANCE + account_str) ##リクエストして結果取得
        if res is None : ##リクエスト失敗判定
          self._fail()
        soup: BeautifulSoup = BeautifulSoup(res.text, 'html.parser') ##beautifulsoupでレスポンス解析
        if soup.title == self.NITTER_ERROR_TITLE: ##タイトルがエラーでないか判定
          print(account_str + "は実在の人物ではありませんでした") ##エラー時ループに戻る
        else:
          print("最終的に出会ったのが@" + account_str + "だった。")
          return account_str ##成功時アカウント名返す

  ##検索クエリを取得
  def _get_query(self, name: Final[str]) -> Response | NoReturn:
    while True:
      print("検索クエリを入れるナリ。複数ある時は一つの塊ごとに入力するナリ。空欄を入れると検索開始ナリ。")
      print("例:「無能」→改行→「脱糞」→改行→「弟殺し」→改行→空欄で改行")
      query_str: list[str] = [name] ##検索クエリ文字列をquery_strに収納。ユーザー名を先に含めておく
      query_input: str = input() ##ユーザー入力受付
      ##空欄が押されるまでユーザー入力受付
      while query_input != '':
        query_str.append(quote(query_input))
        query_input = input()
      res: Response | None = self._request(self.NITTER_INSTANCE + self.SEARCH_QUERY + '+'.join(query_str)) ##リクエストして結果取得
      if res is None : ##リクエスト失敗判定
        self._fail()
      soup: BeautifulSoup = BeautifulSoup(res.text, 'html.parser') ##beautifulsoupでレスポンス解析
      ##1件もないときはHTMLにtimeline-noneがあるから判定
      if soup.find(class_='timeline-none') is None:
        print("クエリのピースが埋まっていく。")
        return res ##成功したリクエストのレスポンスを返す
      else:
        print("適切なクエリを入力することを切に望む。")

  ##接続失敗時処理
  def _fail(self) -> NoReturn:
    print("接続失敗しすぎで強制終了ナリ")
    if self._txt_data != '': ##取得成功したデータがあれば発行
      print("取得成功した分だけ発行しますを")
      self._make_txt()
    exit() ##終了

  ##テキスト発行
  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:
      f.write(result_txt)
    print("テキストファイル手に入ったやで〜")
    exit() ##終了

  ##記録を中断するツイート
  def _stop_word(self) -> str:
    print(f"ここにツイートの本文を入れる!すると・・・・なんと一致するツイートで記録を中断する!(これは本当の仕様です。)ちなみに、空欄だと{self.LIMIT_N_TWEETS}件まで終了しない。")
    end_str: Final[str] = input() ##ユーザー入力受付
    return end_str

  ##Twitterの画像をダウンロード
  def _download_media(self, media_name: Final[str]) -> bool:
    os.makedirs(self.MEDIA_DIR, exist_ok=True)
    url: Final[str] = 'https://pbs.twimg.com/media/' + media_name
    res: Final[Response | None] = self._request(url)
    if res is not None:
      if 'image' not in res.headers['content-type']:
        return False
      with open(self.MEDIA_DIR + '/' + media_name, "wb") as f:
        f.write(res.content)
      return True
    else:
      return False

  def _get_tweet_media(self, tweet: bs4.element.Tag) -> str:
    tweet_media: bs4.element.Tag | None = tweet.select_one('.tweet-body > .attachments') # 引用リツイート内のメディアを選択しないように.tweet-body直下の.attachmentsを選択
    media_txt: str = ''
    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で落ちてきて面倒臭いため取得しない
      """
      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

  def _get_tweet_quote(self, tweet: bs4.element.Tag) -> str:
    tweet_quote: Final[bs4.element.Tag | None] = tweet.select_one('.tweet-body > .quote.quote-big') # 引用リツイートを選択
    quote_txt: str = ''
    if tweet_quote is not None:
      link: str = tweet_quote.select_one('.quote-link').get('href')
      link = re.sub('#.*$', '', link)
      link = self.TWITTER_URL + link
      quote_txt: str = '<br>\n' + self._archive_url(link, link)
    return quote_txt

  #ページからツイート本文をself._txt_dataに収めていく
  def get_tweet(self) -> None | NoReturn:
    soup: Final[BeautifulSoup] = BeautifulSoup(self._page.text, 'html.parser') ##beautifulsoupでレスポンス解析
    tweets: Final[bs4.element.ResultSet] = soup.find_all(class_='timeline-item') ##一ツイートのブロックごとにリストで取得
    for tweet in tweets: ##一ツイート毎に処理
      if tweet.a.text == self.NEWEST: ##Load Newestのボタンは処理しない
        continue
      if tweet.find(class_='retweet-header') is not None: ##retweet-headerはリツイートを示すので入っていれば処理しない
        continue
      tweet_url: str = self.TWITTER_URL + re.sub('#[^#]*$', '', tweet.find(class_='tweet-link').get('href')) ##ツイートのURL作成
      archived_tweet_url: str = self._callinshowlink_url(tweet_url) ##ツイートURLをテンプレートCallinShowlinkに変化
      tweet_content: bs4.element.Tag = tweet.find(class_='tweet-content media-body') ##ツイートの中身だけ取り出す
      self._archive_soup(tweet_content) ##ツイートの中身のリンクをテンプレートArchiveに変化
      media_txt: str = self._get_tweet_media(tweet) ##ツイートに画像などのメディアを追加
      quote_txt: str = self._get_tweet_quote(tweet) ##引用リツイートの場合、元ツイートを追加
      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._limit_count += 1 ##記録回数をカウント
      if self._limit_count % self.REPORT_INTERVAL == 0:
        print(f"ツイートを{self._limit_count}件も記録したンゴwwwwwwwwwww")
      if self._stop != '' and self._stop in tweet_content.get_text(): ##目的ツイートか判定
        print("目的ツイート発見でもう尾張屋根")
        self._make_txt()
      if self._limit_count >= self.LIMIT_N_TWEETS: ##上限達成か判定
        print(f"{self.LIMIT_N_TWEETS}件も記録している。もうやめにしませんか。")
        self._make_txt()

  #tagをテンプレートArchiveの文字列に変化させる
  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:
      if re.match('^https?://', url.get('href')) is not None: ##先頭にhttpが付いていない物はハッシュタグの検索ページへのリンクなので処理しない
        if re.match('https' + self.NITTER_INSTANCE[4:], url.get('href')):
          #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に変化させる
  def _archive_url(self, url: Final[str], text: Final[str]) -> str:
    return '{{Archive|1=' + url + '|2=' + self._archive(url) + '|3=' + text  + '}}' ##テンプレートArchiveの文字列返す

  #URLをテンプレートCallinShowlinkに変化させる
  def _callinshowlink_url(self, url: Final[str]) -> str:
    return '{{CallinShowLink|1=' + url + '|2=' + self._archive(url) + '}}' ##テンプレートCallinShowlinkの文字列返す

  ##URLから魚拓返す
  def _archive(self, url: Final[str]) -> str:
    archive_url: str = self.ARCHIVE_TODAY_STANDARD + re.sub('#', '%23', url) ##wikiに載せるとき用URLで失敗するとこのままhttps://archive.ph/https://xxxxxxxxの形で返される
    res: Final[Response | None] = self._request(self.ARCHIVE_TODAY + re.sub('#', '%23', url)) ##アクセス用URL使って結果を取得
    if res is None : ##魚拓接続失敗時処理
      print(archive_url + 'にアクセス失敗ナリ。出力されるテキストにはそのまま記載されるナリ。')
    else:
      soup: Final[BeautifulSoup] = BeautifulSoup(res.text, 'html.parser') ##beautifulsoupでレスポンス解析
      content: bs4.element.Tag = soup.find(id="CONTENT") ##archive.todayの魚拓一覧ページの中身だけ取得
      if content.get_text()[:len(self.NO_ARCHIVE)] == self.NO_ARCHIVE: ##魚拓があるかないか判定
        print(url + "の魚拓がない。これはいけない。")
      else:
        archive_url = content.find('a').get('href').replace(self.ARCHIVE_TODAY, self.ARCHIVE_TODAY_STANDARD)
    return archive_url

  ##新しいページを取得
  def go_to_new_page(self) -> None | NoReturn:
    soup: Final[BeautifulSoup] = BeautifulSoup(self._page.text, 'html.parser') ##beautifulsoupでレスポンス解析
    show_mores: Final[bs4.element.ResultSet] = soup.find_all(class_="show-more")
    for show_more in show_mores: ##show-moreに次ページへのリンクか前ページへのリンクがある
      if show_more.text != self.NEWEST:  ##前ページへのリンクではないか判定
        new_url: str = self.NITTER_INSTANCE + self.SEARCH + show_more.a.get('href') ##直下のaタグのhrefの中身取ってURL頭部分と合体
    res: Final[Response | None] = self._request(new_url) ##接続してHTML取ってくる
    if res is None:
      self._fail()
    new_page_soup: Final[BeautifulSoup] = BeautifulSoup(res.text, 'html.parser') ##beautifulsoupでレスポンス解析
    if new_page_soup.find(class_="timeline-end") is None: ##ツイートの終端ではtimeline-endだけのページになるので判定
      print(res.url + 'に移動しますを')
      self._page = res ##まだ残りツイートがあるのでページを返して再度ツイート本文収集
    else:
      print("急に残りツイートが無くなったな終了するか")
      self._make_txt()


if __name__ == '__main__':
  krsw: Final[bool] = len(sys.argv) > 1 and sys.argv[1] == 'krsw' ##コマンドライン引数があるかどうかのフラグ
  twitter_archiver: TwitterArchiver = TwitterArchiver(krsw)

  ##ツイートを取得し終えるまでループ
  while True:
    twitter_archiver.get_tweet() ##self._txt_dataにページ内のリツイート以外の全ツイートの中身突っ込んでいく
    twitter_archiver.go_to_new_page() ##新しいページ取得

実行例

20件での実行例。

https://twitter.com/CallinShow/status/1537795802302918656(魚拓)

テレビ局なんか既得権益だろ。

在野の声をSNSを使って行っていくしかない。

https://twitter.com/CallinShow/status/1537796046780497920(魚拓)

オレたちはもう騙されないんだ。

全てを変えるんだ。

https://twitter.com/CallinShow/status/1537796546263416832(魚拓)

なあ、この国は良くなってるのか。

若い人は安く使われ、
物価は上がり、
公共事業だけはお金を垂れ流していく。
少子化が進み、年金も当てにならない。

何でこんなに若い人が苦しまないといけないんだ。

https://twitter.com/CallinShow/status/1537796983578333184(魚拓)

若い人が全員選挙行ったらこの国は変わるぜ。

ガチャのチケットもってるのに使わない手はないだろ。

https://twitter.com/CallinShow/status/1537797605308395520(魚拓)

👍

https://twitter.com/CallinShow/status/1537798386619133952(魚拓)

頼むよ!君にかかってる。

https://twitter.com/CallinShow/status/1537799090326863872(魚拓)

引きこもりの人は引きこもりになりたくてなったんじゃない。
苦しんでるんだ。
誰かがその声を聞く必要がある。
強制的に連れ出すとかやばいだろ。
法規制する必要あるでしょ。
強要罪だぜ。
全ては見過ごされてるんだよ。

https://twitter.com/CallinShow/status/1537803303752388608(魚拓)

この国は

やばいだろ。

引用リスペクト 梅野源治選手

https://twitter.com/CallinShow/status/1538070416488792064(魚拓)

本当、そうだと思う。

貴重な意見ありがとう。

https://twitter.com/CallinShow/status/1538070983827165186(魚拓)

この国は円安が進み、外国旅行に行ったことのない若者も多く。
どうなるんだ。

ライジングサンってどうなってるんだ。

今は戦後の混乱期と同じだ。

若い人たちで国を作るんだ。

https://twitter.com/CallinShow/status/1538071752802467840(魚拓)

立花先生は、いつも一緒懸命だ。

信念は変わらない。

理不尽との闘い。

立花先生が裁判を起こすことは至極まともなことだ。

だって最後は司法なはずだろ。

でも、裁判は、ひどい判決も多い。

権力に擦り寄ることもある。

そんな全てを曝け出しているのが立花先生だ。

こんな人今までいたか。

https://twitter.com/CallinShow/status/1538072997734793216(魚拓)

立花先生はいつも現場主義。

人々の声を拾い。

それを手練手管で伝えいく。

時に道化を案じるのは、みんなに問題を気づいてほしいからだ。

この3年間、少数政党でこれだけ目立ち続けた政党があるか。

それは全て、NHKを変えたい、人々に良くなってもらいたい。

その一心でしかないと思う。

https://twitter.com/CallinShow/status/1538073614540754944(魚拓)

全てをめくっていくんだよ。

ガーシーさんは、その第一章だ。

許せないことを言うことは大切だと。

ガーシー党は、言わば情報公開政党ってことだろ。

いいじゃないか、澱んだ永田町が少しは澄んでくるんじゃないか。

https://twitter.com/CallinShow/status/1538074524721491968(魚拓)

piped.kavin.rocks/nrJqzmMWUCU(魚拓)

澱んだ街ごと食い荒らすんだ。

https://twitter.com/CallinShow/status/1538582552423780352(魚拓)

弁護士唐澤貴洋への匿名のメッセージを募集中!
#マシュマロを投げ合おう
marshmallow-qa.com/apt/21e00…(魚拓)

https://twitter.com/CallinShow/status/1539223436337516548(魚拓)

fsa.go.jp/sesc/news/c_2022/2…(魚拓)

エクシアさん、こういった問題について、どうお考えなんですかね。

https://twitter.com/CallinShow/status/1539223981106302976(魚拓)

エクシアの関戸さんの経営されている会社等のご存じの方いらっしゃいましたらご連絡お願い申し上げます。

https://twitter.com/CallinShow/status/1539224957829672960(魚拓)

そちらは法人名でしょうか?

https://twitter.com/CallinShow/status/1539237355701379075(魚拓)

菊地さんは虎ノ門ヒルズいくらで買ったんだろ。
キャッシュで

https://twitter.com/CallinShow/status/1539237797281816576(魚拓)

仮に3億だとして、これを買うのに、所得としては6億くらいの報酬を得ないとダメだよな。