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

→‎コード: v4.1.13 異常終了時にそれまで収集したツイートを残すように修正
>Fet-Fe
(→‎コード: v4.1.12.post1 docstring修正)
>Fet-Fe
(→‎コード: v4.1.13 異常終了時にそれまで収集したツイートを残すように修正)
11行目: 11行目:
"""Twitter自動収集スクリプト
"""Twitter自動収集スクリプト


ver4.1.12.post1 2024/1/14恒心
ver4.1.13 2024/1/28恒心


当コードは恒心停止してしまった https://rentry.co/7298g の降臨ショーツイート自動収集スクリプトの復刻改善版です
当コードは恒心停止してしまった https://rentry.co/7298g の降臨ショーツイート自動収集スクリプトの復刻改善版です
64行目: 64行目:
import re
import re
import shutil
import shutil
import signal
import subprocess
import subprocess
import sys
import sys
76行目: 77行目:
from time import sleep
from time import sleep
from traceback import TracebackException
from traceback import TracebackException
from types import MappingProxyType, TracebackType
from types import FrameType, MappingProxyType, TracebackType
from typing import (Final, NamedTuple, NoReturn, Self, assert_never, final,
from typing import (Final, NamedTuple, NoReturn, Self, assert_never, final,
                     override)
                     override)
125行目: 126行目:
         """
         """


         nitter_instance: Final[str] = 'https://nitter.moomoo.me/'
         nitter_instance: Final[str] = 'https://nitter.unixfox.eu/'
         """Final[str]: Nitterのインスタンス。
         """Final[str]: Nitterのインスタンス。


329行目: 330行目:
             return None
             return None


     def _choose_tor_proxies(self) -> dict[str, str] | None | NoReturn:
     def _choose_tor_proxies(self) -> dict[str, str] | None:
         """Torを使うのに必要なプロキシ情報を返す。
         """Torを使うのに必要なプロキシ情報を返す。


341行目: 342行目:


         Returns:
         Returns:
             dict[str, str] | None | NoReturn: プロキシ情報。
             dict[str, str] | None: プロキシ情報。
         """
         """
         logger.info('Torのチェック中ですを')
         logger.info('Torのチェック中ですを')
1,122行目: 1,123行目:
         return True
         return True


     def _check_constants(self) -> None | NoReturn:
     def _check_constants(self) -> None:
         """定数のバリデーションを行う。
         """定数のバリデーションを行う。


         Returns:
         Returns:
             None | NoReturn: すべての対象定数が正しければ `None`。失敗したら例外を出す。
             None: すべての対象定数が正しければ `None`。失敗したら例外を出す。


         Raises:
         Raises:
1,145行目: 1,146行目:
         return shutil.which('ffmpeg') is not None
         return shutil.which('ffmpeg') is not None


     def _check_nitter_instance(
     def _check_nitter_instance(self, accessor: AccessorHandler) -> None:
            self, accessor: AccessorHandler) -> None | NoReturn:
         """Nitterのインスタンスが生きているかチェックする。
         """Nitterのインスタンスが生きているかチェックする。


1,157行目: 1,157行目:


         Returns:
         Returns:
             None | NoReturn: Nitterにアクセスできれば `None`。できなければ終了。
             None: Nitterにアクセスできれば `None`。できなければ終了。
         """
         """
         logger.info('Nitterのインスタンスチェック中ですを')
         logger.info('Nitterのインスタンスチェック中ですを')
1,168行目: 1,168行目:
         logger.info('Nitter OK')
         logger.info('Nitter OK')


     def _check_archive_instance(
     def _check_archive_instance(self, accessor: AccessorHandler) -> None:
            self, accessor: AccessorHandler) -> None | NoReturn:
         """archive.todayのTor用インスタンスが生きているかチェックする。
         """archive.todayのTor用インスタンスが生きているかチェックする。


1,176行目: 1,175行目:


         Returns:
         Returns:
             None | NoReturn: archive.todayのTorインスタンスにアクセスできれば `None`。できなければ終了。
             None: archive.todayのTorインスタンスにアクセスできれば `None`。できなければ終了。
         """
         """
         logger.info('archive.todayのTorインスタンスチェック中ですを')
         logger.info('archive.todayのTorインスタンスチェック中ですを')
1,188行目: 1,187行目:


     def _invidious_instances(
     def _invidious_instances(
             self,
             self, accessor: AccessorHandler) -> tuple[str, ...]:
            accessor: AccessorHandler) -> tuple[str, ...] | NoReturn:
         """Invidiousのインスタンスのタプルを取得する。
         """Invidiousのインスタンスのタプルを取得する。


1,196行目: 1,194行目:


         Returns:
         Returns:
             tuple[str, ...] | NoReturn: Invidiousのインスタンスのタプル。インスタンスが死んでいれば終了。
             tuple[str, ...]: Invidiousのインスタンスのタプル。インスタンスが死んでいれば終了。
         """
         """
         logger.info('Invidiousのインスタンスリストを取得中ですを')
         logger.info('Invidiousのインスタンスリストを取得中ですを')
1,549行目: 1,547行目:
         return poll_txt
         return poll_txt


     def _get_timeline_items(self, soup: BeautifulSoup) -> list[Tag]:
     def _get_timeline_items(self, soup: Tag) -> list[Tag]:
         """タイムラインのツイートを取得。
         """タイムラインのツイートを取得。


1,555行目: 1,553行目:


         Args:
         Args:
             soup (BeautifulSoup): Nitterのページを表すBeautifulSoupオブジェクト。
             soup (Tag): Nitterのページを表すbeautifulsoup4タグ。


         Returns:
         Returns:
1,835行目: 1,833行目:


     def execute(self, krsw: bool = False, use_browser: bool = True,
     def execute(self, krsw: bool = False, use_browser: bool = True,
                 enable_javascript: bool = True) -> None | NoReturn:
                 enable_javascript: bool = True) -> None:
         """通信が必要な部分のロジック。
         """通信が必要な部分のロジック。


1,846行目: 1,844行目:
                 `True`。
                 `True`。
         """
         """
        def signal_handler(signum: int, frame: FrameType | None) -> NoReturn:
            """ユーザがCtrl + Cでプログラムを止めたときのシグナルハンドラ。
            Args:
                signum (int): シグナル番号。想定されるのはSIGINTのみ。
                frame (FrameType | None): 現在のスタックフレーム。
            """
            logger.info('ユーザがプログラムを中止したなりを')
            self._table_builder.dump_file()
            sys.exit()
         # Seleniumドライバーを必ず終了するため、with文を利用する。
         # Seleniumドライバーを必ず終了するため、with文を利用する。
         with AccessorHandler(use_browser, enable_javascript) as accessor:
         with AccessorHandler(use_browser, enable_javascript) as accessor:
1,863行目: 1,872行目:


             # ツイートを取得し終えるまでループ
             # ツイートを取得し終えるまでループ
             while True:
             signal.signal(signal.SIGINT, signal_handler)
                if not self._get_tweet(accessor):
            try:
                    break
                while True:
                if not self._go_to_new_page(accessor):
                    if not self._get_tweet(accessor):
                    break
                        break
                    if not self._go_to_new_page(accessor):
                        break
            except BaseException as e:
                logger.critical('予想外のエラーナリ。ここまでの成果をダンプして終了するナリ。')
                self._table_builder.dump_file()
                raise e




1,956行目: 1,971行目:


     @override
     @override
     def _check_constants(self) -> None | NoReturn:
     def _check_constants(self) -> None:
         super()._check_constants()
         super()._check_constants()
         assert (isinstance(self.INCREMENTED_NUM_DEFAULT, int)
         assert (isinstance(self.INCREMENTED_NUM_DEFAULT, int)
2,006行目: 2,021行目:
                     self._url_list_on_wiki.append(href)
                     self._url_list_on_wiki.append(href)


     def _append_tweet_urls(self, soup: BeautifulSoup) -> None:
     def _append_tweet_urls(self, soup: Tag) -> None:
         """ツイートのURLを保存する。
         """ツイートのURLを保存する。


         Args:
         Args:
             soup (BeautifulSoup): archive.todayでのURL検索結果のページのオブジェクト。
             soup (Tag): archive.todayでのURL検索結果のページのオブジェクト。
         """
         """
         tweets: Final[ResultSet[Tag]] = soup.select(
         tweets: Final[ResultSet[Tag]] = soup.select(
2,043行目: 2,058行目:


     def _fetch_next_page(
     def _fetch_next_page(
             self,
             self, soup: Tag, accessor: AccessorHandler) -> str | None:
            soup: BeautifulSoup,
            accessor: AccessorHandler) -> str | None:
         """archive.todayの検索結果のページをpaginateする。
         """archive.todayの検索結果のページをpaginateする。


         Args:
         Args:
             soup (BeautifulSoup): archive.todayでのURL検索結果のページのオブジェクト。
             soup (Tag): archive.todayでのURL検索結果のページのオブジェクト。
             accessor (AccessorHandler): アクセスハンドラ。
             accessor (AccessorHandler): アクセスハンドラ。


2,065行目: 2,078行目:
             return
             return


     def _get_tweet_loop(
     def _get_tweet_loop(self, soup: Tag, accessor: AccessorHandler) -> None:
            self,
            soup: BeautifulSoup,
            accessor: AccessorHandler) -> None:
         """archive.todayの検索結果に対して、paginateしながら未記載のツイートURLを記録する。
         """archive.todayの検索結果に対して、paginateしながら未記載のツイートURLを記録する。


         Args:
         Args:
             soup (BeautifulSoup): archive.todayでのURL検索結果のページのオブジェクト。
             soup (Tag): archive.todayでのURL検索結果のページのオブジェクト。
             accessor (AccessorHandler): アクセスハンドラ。
             accessor (AccessorHandler): アクセスハンドラ。
         """
         """
2,366行目: 2,376行目:
     @override
     @override
     def execute(self, krsw: bool = False, use_browser: bool = True,
     def execute(self, krsw: bool = False, use_browser: bool = True,
                 enable_javascript: bool = True) -> None | NoReturn:
                 enable_javascript: bool = True) -> None:
         logger.info('Wikiに未掲載のツイートのURLを収集しますを')
         logger.info('Wikiに未掲載のツイートのURLを収集しますを')
         warnings.warn(
         warnings.warn(
匿名利用者