「利用者:夜泣き/スクリプト」の版間の差分
ナビゲーションに移動
検索に移動
→コード: v4.1.13 異常終了時にそれまで収集したツイートを残すように修正
>Fet-Fe 細 (→コード: v4.1.12.post1 docstring修正) |
>Fet-Fe (→コード: v4.1.13 異常終了時にそれまで収集したツイートを残すように修正) |
||
11行目: | 11行目: | ||
"""Twitter自動収集スクリプト | """Twitter自動収集スクリプト | ||
ver4.1. | 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. | 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 | def _choose_tor_proxies(self) -> dict[str, str] | None: | ||
"""Torを使うのに必要なプロキシ情報を返す。 | """Torを使うのに必要なプロキシ情報を返す。 | ||
341行目: | 342行目: | ||
Returns: | Returns: | ||
dict[str, str] | None | dict[str, str] | None: プロキシ情報。 | ||
""" | """ | ||
logger.info('Torのチェック中ですを') | logger.info('Torのチェック中ですを') | ||
1,122行目: | 1,123行目: | ||
return True | return True | ||
def _check_constants(self) -> None | def _check_constants(self) -> None: | ||
"""定数のバリデーションを行う。 | """定数のバリデーションを行う。 | ||
Returns: | Returns: | ||
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: | ||
"""Nitterのインスタンスが生きているかチェックする。 | """Nitterのインスタンスが生きているかチェックする。 | ||
1,157行目: | 1,157行目: | ||
Returns: | Returns: | ||
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: | ||
"""archive.todayのTor用インスタンスが生きているかチェックする。 | """archive.todayのTor用インスタンスが生きているかチェックする。 | ||
1,176行目: | 1,175行目: | ||
Returns: | Returns: | ||
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, ...]: | ||
"""Invidiousのインスタンスのタプルを取得する。 | """Invidiousのインスタンスのタプルを取得する。 | ||
1,196行目: | 1,194行目: | ||
Returns: | Returns: | ||
tuple[str, ...] | 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: | def _get_timeline_items(self, soup: Tag) -> list[Tag]: | ||
"""タイムラインのツイートを取得。 | """タイムラインのツイートを取得。 | ||
1,555行目: | 1,553行目: | ||
Args: | Args: | ||
soup ( | 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 | 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) | ||
try: | |||
while True: | |||
if not self._get_tweet(accessor): | |||
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 | 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: | def _append_tweet_urls(self, soup: Tag) -> None: | ||
"""ツイートのURLを保存する。 | """ツイートのURLを保存する。 | ||
Args: | Args: | ||
soup ( | 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: | ||
"""archive.todayの検索結果のページをpaginateする。 | """archive.todayの検索結果のページをpaginateする。 | ||
Args: | Args: | ||
soup ( | 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: | ||
"""archive.todayの検索結果に対して、paginateしながら未記載のツイートURLを記録する。 | """archive.todayの検索結果に対して、paginateしながら未記載のツイートURLを記録する。 | ||
Args: | Args: | ||
soup ( | 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 | enable_javascript: bool = True) -> None: | ||
logger.info('Wikiに未掲載のツイートのURLを収集しますを') | logger.info('Wikiに未掲載のツイートのURLを収集しますを') | ||
warnings.warn( | warnings.warn( |