→コード: v3.0.3 機能に変化なし。機能の説明コメントをGoogleスタイルのdocstringに変更
>Fet-Fe (→コード: v3.0.2 .timeline-item.unavailableは無視する) |
>Fet-Fe (→コード: v3.0.3 機能に変化なし。機能の説明コメントをGoogleスタイルのdocstringに変更) |
||
5行目: | 5行目: | ||
#!/usr/bin/env python3 | #!/usr/bin/env python3 | ||
"""Twitter自動収集スクリプト | |||
ver3.0.3 2023/7/16恒心 | |||
当コードは恒心停止してしまった https://rentry.co/7298g の降臨ショーツイート自動収集スクリプトの復刻改善版です | |||
前開発者との出会いに感謝 | 前開発者との出会いに感謝 | ||
Examples: | |||
定数類は状況に応じて変えてください。 | |||
:: | |||
$ python3 (ファイル名) | |||
コマンドライン引数に ``krsw`` とつけると自動モードになります。 | |||
:: | |||
$ python3 (ファイル名) krsw | |||
自動モードではユーザーは降臨ショー、クエリはなし、取得上限まで自動です。 | |||
つまりユーザー入力が要りません。 | |||
Note: | |||
* Pythonのバージョンは3.10以上 | |||
* 環境は玉葱前提です。 | |||
* Whonix-Workstation, MacOSで動作確認済 | |||
* MacOSの場合はbrewでtorコマンドを導入し、実行 | |||
* PySocks, bs4はインストールしないと標準で入ってません | |||
* requestsも環境によっては入っていないかもしれない | |||
* $ pip install bs4 requests PySocks | |||
* pipも入っていなければ ``$ sudo apt install pip`` | |||
* `ffmpeg <https://ffmpeg.org>`_ が入っていると動画も自動取得しますが、無くても動きます | |||
* バグ報告は `利用者・トーク:夜泣き#利用者:夜泣き/スクリプトについて <https://krsw-wiki.org/wiki/利用者・トーク:夜泣き#利用者:夜泣き/スクリプトについて>`_ | |||
""" | |||
#インポート類 | #インポート類 | ||
58行目: | 64行目: | ||
warnings.simplefilter('ignore') | warnings.simplefilter('ignore') | ||
Response: TypeAlias = requests.models.Response | Response: TypeAlias = requests.models.Response | ||
"""TypeAlias: requests.models.Responseの型エイリアス。 | |||
""" | |||
class TwitterArchiver: | class TwitterArchiver: | ||
"""ツイートをWikiの形式にダンプするクラス。 | |||
Nitterからツイートを取得し、Wikiの形式にダンプする。 | |||
削除されたツイートや編集前のツイートは取得できない。 | |||
""" | |||
#定数・設定類 | #定数・設定類 | ||
NITTER_INSTANCE: Final[str] = 'http://nitter7bryz3jv7e3uekphigvmoyoem4al3fynerxkj22dmoxoq553qd.onion/' | NITTER_INSTANCE: Final[str] = 'http://nitter7bryz3jv7e3uekphigvmoyoem4al3fynerxkj22dmoxoq553qd.onion/' | ||
"""Final[str]: Nitterのインスタンス。 | |||
生きているのは https://github.com/zedeus/nitter/wiki/Instances で確認。 | |||
Note: | |||
末尾にスラッシュ必須。 | |||
""" | |||
ARCHIVE_TODAY: Final[str] = 'http://archiveiya74codqgiixo33q62qlrqtkgmcitqx5u2oeqnmn5bpcbiyd.onion/' | ARCHIVE_TODAY: Final[str] = 'http://archiveiya74codqgiixo33q62qlrqtkgmcitqx5u2oeqnmn5bpcbiyd.onion/' | ||
"""Final[str]: archive.todayの魚拓のonionドメイン。 | |||
ページにアクセスして魚拓があるか調べるのにはonion版のarchive.todayを使用。 | |||
Note: | |||
末尾にスラッシュ必須。 | |||
""" | |||
ARCHIVE_TODAY_STANDARD: Final[str] = 'https://archive.vn/' | ARCHIVE_TODAY_STANDARD: Final[str] = 'https://archive.vn/' | ||
"""Final[str]: archive.todayの魚拓のクリアネットドメイン。 | |||
記事にはクリアネット用のarchive.todayリンクを貼る。 | |||
Note: | |||
末尾にスラッシュ必須。 | |||
""" | |||
TWITTER_URL: Final[str] = 'https://twitter.com/' | TWITTER_URL: Final[str] = 'https://twitter.com/' | ||
"""Final[str]: TwitterのURL。 | |||
Note: | |||
末尾にスラッシュ必須。 | |||
""" | |||
CALLINSHOW: Final[str] = 'CallinShow' | CALLINSHOW: Final[str] = 'CallinShow' | ||
"""Final[str]: 降臨ショーのユーザーネーム。 | |||
""" | |||
REQUEST_TIMEOUT: Final[int] = 30 | REQUEST_TIMEOUT: Final[int] = 30 | ||
"""Final[int]: HTTPリクエストのタイムアウト秒数。 | |||
""" | |||
LIMIT_N_TWEETS: Final[int] = 100 | LIMIT_N_TWEETS: Final[int] = 100 | ||
"""Final[int]: 取得するツイート数の上限。 | |||
""" | |||
REPORT_INTERVAL: Final[int] = 5 | REPORT_INTERVAL: Final[int] = 5 | ||
"""Final[int]: 記録件数を報告するインターバル。 | |||
""" | |||
LIMIT_N_REQUESTS: Final[int] = 5 | LIMIT_N_REQUESTS: Final[int] = 5 | ||
"""Final[int]: HTTPリクエスト失敗時の再試行回数。 | |||
""" | |||
WAIT_TIME_FOR_ERROR: Final[int] = 4 | WAIT_TIME_FOR_ERROR: Final[int] = 4 | ||
"""Final[int]: HTTPリクエスト失敗時にさらに追加する待機時間。 | |||
""" | |||
WAIT_TIME: Final[int] = 1 | WAIT_TIME: Final[int] = 1 | ||
"""Final[int]: HTTPリクエスト成功失敗関わらず待機時間。 | |||
1秒待つだけで行儀がいいクローラーだそうなので既定では1秒。 | |||
しかし日本のポリホーモは1秒待っていても捕まえてくるので注意。 | |||
https://ja.wikipedia.org/wiki/?curid=2187212 | |||
""" | |||
TWEETS_OR_REPLIES: Final[str] = 'with_replies' | TWEETS_OR_REPLIES: Final[str] = 'with_replies' | ||
"""Final[str]: NitterのURLのドメインとユーザーネーム部分の接続部品。 | |||
""" | |||
HEADERS: Final[dict[str, str]] = { | HEADERS: Final[dict[str, str]] = { | ||
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; rv:91.0) Gecko/20100101 Firefox/91.0' | 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; rv:91.0) Gecko/20100101 Firefox/91.0' | ||
} | } | ||
"""Final[dict[str, str]]: HTTPリクエスト時のユーザーエージェントとヘッダ。 | |||
""" | |||
PROXIES: Final[dict[str, str]] = { | PROXIES: Final[dict[str, str]] = { | ||
'http': 'socks5h://127.0.0.1:9050', | 'http': 'socks5h://127.0.0.1:9050', | ||
'https': 'socks5h://127.0.0.1:9050' | 'https': 'socks5h://127.0.0.1:9050' | ||
} | } | ||
"""Final[dict[str, str]]: HTTPプロキシの設定。 | |||
""" | |||
NITTER_ERROR_TITLE: Final[str] = 'Error|nitter' | NITTER_ERROR_TITLE: Final[str] = 'Error|nitter' | ||
"""Final[str]: Nitterでユーザーがいなかったとき返ってくるページのタイトル。 | |||
万が一仕様変更で変わったとき用。 | |||
""" | |||
NO_ARCHIVE: Final[str] = 'No results' | NO_ARCHIVE: Final[str] = 'No results' | ||
"""Final[str]: archive.todayで魚拓がなかったときのレスポンス。 | |||
万が一仕様変更で変わったとき用。 | |||
""" | |||
NEWEST: Final[str] = 'Load newest' | NEWEST: Final[str] = 'Load newest' | ||
"""Final[str]: Nitterの前ページ読み込み部分の名前。 | |||
万が一仕様変更で変わったとき用。 | |||
""" | |||
MEDIA_DIR: Final[str] = 'tweet_media' | MEDIA_DIR: Final[str] = 'tweet_media' | ||
"""Final[str]: 画像などのツイートメディアをダウンロードするディレクトリ。 | |||
""" | |||
def __init__(self, krsw: bool=False): | def __init__(self, krsw: bool=False): | ||
"""コンストラクタ | |||
:class:`~TwitterArchiver` のメンバをセットし、ツイートを収集するアカウントと | |||
検索クエリ、終わりにするツイートを入力させる。 | |||
Args: | |||
krsw bool: Trueの場合、名前が自動で :const:`~CALLINSHOW` になり、クエリと終わりにするツイートが自動で無しになる。 | |||
""" | |||
self._txt_data: list[str] = [] | self._txt_data: list[str] = [] | ||
self._limit_count: int = 0 ##記録数 | self._limit_count: int = 0 ##記録数 | ||
179行目: | 232行目: | ||
print() | print() | ||
def _request_once(self, url: Final[str]) -> Response: | |||
"""引数のURLにHTTP接続します。 | |||
Args: | |||
url Final[str]: 接続するURL | |||
Returns: | |||
Response: レスポンス | |||
Note: | |||
失敗かどうかは呼出側で要判定 | |||
""" | |||
if self._proxy_is_needed: | if self._proxy_is_needed: | ||
res: Response = 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) | ||
194行目: | 251行目: | ||
return res | return res | ||
def _request(self, url: Final[str]) -> Response | None: | def _request(self, url: Final[str]) -> Response | None: | ||
"""HTTP接続を :const:`~LIMIT_N_REQUESTS` で定義された回数まで試します。 | |||
成功すると結果を返します。 | |||
接続失敗が何度も起きるとNoneを返します。 | |||
Args: | |||
url Final[str]: 接続するURL | |||
Returns: | |||
Response | None: レスポンス。接続失敗が何度も起きるとNoneを返します。 | |||
Note: | |||
失敗かどうかは呼出側で要判定 | |||
""" | |||
counter: int = 1 ##リクエスト挑戦回数を記録 | counter: int = 1 ##リクエスト挑戦回数を記録 | ||
while True: | while True: | ||
214行目: | 281行目: | ||
return res ##リクエストの結果返す | return res ##リクエストの結果返す | ||
def _check_slash(self) -> None | NoReturn: | def _check_slash(self) -> None | NoReturn: | ||
"""URLの最後にスラッシュが付いていなければエラーを出します。 | |||
Returns: | |||
None | NoReturn: すべてのURLが正しければNone。失敗したら例外を出す。 | |||
Raises: | |||
RuntimeError: URLの最後にスラッシュがついていない場合に出る。 | |||
""" | |||
if self.NITTER_INSTANCE[-1] != '/': | if self.NITTER_INSTANCE[-1] != '/': | ||
raise RuntimeError('NITTER_INSTANCEの末尾には/が必須です') | raise RuntimeError('NITTER_INSTANCEの末尾には/が必須です') | ||
225行目: | 299行目: | ||
raise RuntimeError('TWITTER_URLの末尾には/が必須です') | raise RuntimeError('TWITTER_URLの末尾には/が必須です') | ||
def _check_tor_proxy_is_needed(self) -> bool | NoReturn: | def _check_tor_proxy_is_needed(self) -> bool | NoReturn: | ||
"""Torが使えているかチェックします。 | |||
Returns: | |||
bool | NoReturn: Tor用のプロキシを通さなくてもTor通信になっていればFalse。プロキシを通す必要があればTrue。 | |||
Raises: | |||
RuntimeError: https://check.torproject.org/api/ip にアクセスできなければ出る。 | |||
""" | |||
initial_proxy_is_needed: bool = self._proxy_is_needed | initial_proxy_is_needed: bool = self._proxy_is_needed | ||
print('Torのチェック中ですを') | print('Torのチェック中ですを') | ||
254行目: | 335行目: | ||
exit(1) | exit(1) | ||
def _check_nitter_instance(self) -> None | NoReturn: | def _check_nitter_instance(self) -> None | NoReturn: | ||
"""Nitterのインスタンスが生きているかチェックする。 | |||
死んでいたらそこで終了。 | |||
接続を一回しか試さない :func:`~_request_once` を使っているのは、激重インスタンスが指定されたとき試行回数増やして偶然成功してそのまま実行されるのを躱すため。 | |||
Returns: | |||
None | NoReturn: NitterにアクセスできればNone。できなければ終了。 | |||
""" | |||
print("Nitterのインスタンスチェック中ですを") | print("Nitterのインスタンスチェック中ですを") | ||
try: | try: | ||
269行目: | 354行目: | ||
print("Nitter OK") | print("Nitter OK") | ||
def _check_archive_instance(self) -> None | NoReturn: | def _check_archive_instance(self) -> None | NoReturn: | ||
"""archive.todayのTor用インスタンスが生きているかチェックする。 | |||
Returns: | |||
None | NoReturn: archive.todayのTorインスタンスにアクセスできればNone。できなければ終了。 | |||
""" | |||
print("archive.todayのTorインスタンスチェック中ですを") | print("archive.todayのTorインスタンスチェック中ですを") | ||
try: | try: | ||
281行目: | 370行目: | ||
print("archive.today OK") | print("archive.today OK") | ||
def _invidious_instances(self) -> tuple[str] | NoReturn: | def _invidious_instances(self) -> tuple[str] | NoReturn: | ||
"""Invidiousのインスタンスのタプルを取得する。 | |||
Returns: | |||
tuple[str] | NoReturn: Invidiousのインスタンスのタプル。Invidiousのインスタンスが死んでいれば終了。 | |||
""" | |||
print("Invidiousのインスタンスリストを取得中ですを") | print("Invidiousのインスタンスリストを取得中ですを") | ||
invidious_json: Response | None = self._request('https://api.invidious.io/instances.json') | invidious_json: Response | None = self._request('https://api.invidious.io/instances.json') | ||
300行目: | 393行目: | ||
def _get_name(self) -> str | NoReturn: | def _get_name(self) -> str | NoReturn: | ||
"""ツイート収集するユーザー名を標準入力から取得する。 | |||
何も入力しないと :const:`~CALLINSHOW` を指定する。 | |||
Returns: | |||
str | NoReturn: ユーザ名。ユーザページの取得に失敗したら終了。 | |||
""" | |||
while True: | while True: | ||
print('アカウント名を入れなければない。空白だと自動的に' + self.CALLINSHOW + 'になりますを') | print('アカウント名を入れなければない。空白だと自動的に' + self.CALLINSHOW + 'になりますを') | ||
320行目: | 418行目: | ||
return account_str ##成功時アカウント名返す | return account_str ##成功時アカウント名返す | ||
def _get_query(self) -> None: | def _get_query(self) -> None: | ||
"""検索クエリを標準入力から取得する。 | |||
取得したクエリは ``self.query_strs`` に加えられる。 | |||
""" | |||
print("検索クエリを入れるナリ。複数ある時は一つの塊ごとに入力するナリ。空欄を入れると検索開始ナリ。") | print("検索クエリを入れるナリ。複数ある時は一つの塊ごとに入力するナリ。空欄を入れると検索開始ナリ。") | ||
print("例:「無能」→改行→「脱糞」→改行→「弟殺し」→改行→空欄で改行") | print("例:「無能」→改行→「脱糞」→改行→「弟殺し」→改行→空欄で改行") | ||
331行目: | 432行目: | ||
print("クエリのピースが埋まっていく。") | print("クエリのピースが埋まっていく。") | ||
def _fail(self) -> NoReturn: | def _fail(self) -> NoReturn: | ||
"""接続失敗時処理。 | |||
取得に成功した分だけファイルにダンプし、プログラムを終了する。 | |||
""" | |||
print("接続失敗しすぎで強制終了ナリ") | print("接続失敗しすぎで強制終了ナリ") | ||
if len(self._txt_data) > 0: ##取得成功したデータがあれば発行 | if len(self._txt_data) > 0: ##取得成功したデータがあれば発行 | ||
340行目: | 444行目: | ||
exit(1) ##終了 | exit(1) ##終了 | ||
def _convert_to_text_table(self, text: str) -> str: | |||
def _convert_to_text_table(self, text) -> str: | """``self._txt_data[0]`` にwikiでテーブル表示にするためのヘッダとフッタをつける。 | ||
Args: | |||
text str: ヘッダとフッタがないWikiテーブル。 | |||
Returns: | |||
str: テーブル表示用のヘッダとフッタがついたWikiテーブル。 | |||
""" | |||
return '{|class="wikitable" style="text-align: left;"\n' + text + '|}' | return '{|class="wikitable" style="text-align: left;"\n' + text + '|}' | ||
def _make_txt(self) -> NoReturn: | def _make_txt(self) -> NoReturn: | ||
"""Wikiテーブルをファイル出力し、プログラムを終了する。 | |||
""" | |||
self._next_day() | self._next_day() | ||
result_txt: Final[str] = '\n'.join(self._txt_data) ##リストを合体 | result_txt: Final[str] = '\n'.join(self._txt_data) ##リストを合体 | ||
354行目: | 466行目: | ||
exit(0) ##終了 | exit(0) ##終了 | ||
def _stop_word(self) -> str: | def _stop_word(self) -> str: | ||
"""ツイートの記録を中断するための文をユーザに入力させる。 | |||
Returns: | |||
str: ツイートの記録を中断するための文。ツイート内容の一文がその文と一致したら記録を中断する。 | |||
""" | |||
print(f"ここにツイートの本文を入れる!すると・・・・なんとそれを含むツイートで記録を中断する!(これは本当の仕様です。)ちなみに、空欄だと{self.LIMIT_N_TWEETS}件まで終了しない。") | print(f"ここにツイートの本文を入れる!すると・・・・なんとそれを含むツイートで記録を中断する!(これは本当の仕様です。)ちなみに、空欄だと{self.LIMIT_N_TWEETS}件まで終了しない。") | ||
end_str: Final[str] = input() ##ユーザー入力受付 | end_str: Final[str] = input() ##ユーザー入力受付 | ||
return end_str | return end_str | ||
def _download_media(self, media_name: Final[str]) -> bool: | def _download_media(self, media_name: Final[str]) -> bool: | ||
"""ツイートの画像をダウンロードし、:const:`~MEDIA_DIR` に保存する。 | |||
Args: | |||
media_name Final[str]: 画像ファイル名。Nitter上のimgタグのsrc属性では、``/pic/media%2F`` に後続する。 | |||
Returns: | |||
bool: 保存に成功したかどうか。 | |||
""" | |||
os.makedirs(self.MEDIA_DIR, exist_ok=True) | os.makedirs(self.MEDIA_DIR, exist_ok=True) | ||
url: Final[str] = urljoin('https://pbs.twimg.com/media/', media_name) | url: Final[str] = urljoin('https://pbs.twimg.com/media/', media_name) | ||
374行目: | 497行目: | ||
return False | return False | ||
def _tweet_date(self, tweet: bs4.element.Tag) -> datetime: | def _tweet_date(self, tweet: bs4.element.Tag) -> datetime: | ||
"""ツイートの時刻を取得する。 | |||
Args: | |||
tweet bs4.element.Tag: ツイート内容である ``.timeline-item`` タグを表すbeautifulsoup4タグ。 | |||
Returns: | |||
datetime: ツイートの時刻。 | |||
""" | |||
date_str: str = tweet.find(class_='tweet-date').a['title'] | date_str: str = tweet.find(class_='tweet-date').a['title'] | ||
date: datetime = datetime.strptime(date_str, '%b %d, %Y · %I:%M %p %Z').replace(tzinfo=ZoneInfo('UTC')).astimezone(ZoneInfo('Asia/Tokyo')) | date: datetime = datetime.strptime(date_str, '%b %d, %Y · %I:%M %p %Z').replace(tzinfo=ZoneInfo('UTC')).astimezone(ZoneInfo('Asia/Tokyo')) | ||
381行目: | 511行目: | ||
#self._dateの日付のツイートがなくなったときの処理 | #self._dateの日付のツイートがなくなったときの処理 | ||
def _next_day(self, date: datetime | None = None) -> None: | def _next_day(self, date: datetime|None = None) -> None: | ||
"""1日分のツイートをテーブル形式に変換し、その日のツイートを記録し終わったことを通知して、``self._txt_data`` の0番目に空文字列を追加する。 | |||
Args: | |||
date datetime|None: | |||
記録した日付の前日の日付。Noneでなければ、``self._date`` をその値に更新する。 | |||
""" | |||
if self._txt_data[0]: # 空でなければ出力 | if self._txt_data[0]: # 空でなければ出力 | ||
self._txt_data[0] = self._convert_to_text_table(self._txt_data[0]) | self._txt_data[0] = self._convert_to_text_table(self._txt_data[0]) | ||
395行目: | 531行目: | ||
def _get_tweet_media(self, tweet: bs4.element.Tag) -> str: | def _get_tweet_media(self, tweet: bs4.element.Tag) -> str: | ||
"""ツイートの画像や動画を取得する。 | |||
Args: | |||
tweet bs4.element.Tag: ツイート内容である ``.timeline-item`` タグを表すbeautifulsoup4タグ。 | |||
Returns: | |||
str: Wiki記法でのファイルへのリンクの文字列。 | |||
""" | |||
tweet_media: bs4.element.Tag | None = 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: str = '' | media_txt: str = '' | ||
441行目: | 585行目: | ||
def _get_tweet_quote(self, tweet: bs4.element.Tag) -> str: | def _get_tweet_quote(self, tweet: bs4.element.Tag) -> str: | ||
"""引用リツイートの引用元へのリンクを取得する。 | |||
Args: | |||
tweet bs4.element.Tag: ツイート内容である ``.timeline-item`` タグを表すbeautifulsoup4タグ。 | |||
Returns: | |||
str: Archiveテンプレートでラップされた引用元ツイートへのリンク。 | |||
""" | |||
tweet_quote: Final[bs4.element.Tag | None] = 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: str = '' | quote_txt: str = '' | ||
454行目: | 606行目: | ||
def _get_tweet_poll(self, tweet: bs4.element.Tag) -> str: | def _get_tweet_poll(self, tweet: bs4.element.Tag) -> str: | ||
"""ツイートの投票結果を取得する。 | |||
Args: | |||
tweet bs4.element.Tag: ツイート内容である ``.timeline-item`` タグを表すbeautifulsoup4タグ。 | |||
Returns: | |||
str: Wiki形式に書き直した投票結果。 | |||
""" | |||
tweet_poll: Final[bs4.element.Tag | None] = tweet.select_one('.tweet-body > .poll') | tweet_poll: Final[bs4.element.Tag | None] = tweet.select_one('.tweet-body > .poll') | ||
poll_txt: str = '' | poll_txt: str = '' | ||
467行目: | 627行目: | ||
return poll_txt | return poll_txt | ||
def _get_timeline_items(self, soup: BeautifulSoup) -> list[bs4.element.Tag]: | def _get_timeline_items(self, soup: BeautifulSoup) -> list[bs4.element.Tag]: | ||
"""タイムラインのツイートを取得。 | |||
基本的に投稿時刻の降順に取得し、リプライツリーは最後のツイートの時刻を基準として降順にひとまとまりにする。 | |||
Args: | |||
soup BeautifulSoup: Nitterのページを表すBeautifulSoupオブジェクト。 | |||
Returns: | |||
list[bs4.element.Tag]: ツイートのアイテムである ``.timeline-item`` タグを表すbs4.element.Tagオブジェクトのリスト。 | |||
""" | |||
timeline_item_list: list[bs4.element.Tag] = [] | timeline_item_list: list[bs4.element.Tag] = [] | ||
for item_or_list in soup.select('.timeline > .timeline-item, .timeline > .thread-line'): | for item_or_list in soup.select('.timeline > .timeline-item, .timeline > .thread-line'): | ||
474行目: | 643行目: | ||
continue | continue | ||
elif 'thread-line' in item_or_list.attrs['class']: | elif 'thread-line' in item_or_list.attrs['class']: | ||
# そのままtimeline-itemクラスをfind_allするとツイートの順番が逆転するので、順番通りに取得するよう処理 | |||
for item in reversed(item_or_list.select('.timeline-item')): | for item in reversed(item_or_list.select('.timeline-item')): | ||
timeline_item_list.append(item) | timeline_item_list.append(item) | ||
481行目: | 651行目: | ||
def get_tweet(self) -> None | NoReturn: | def get_tweet(self) -> None | NoReturn: | ||
"""ページからツイート本文を ``self._txt_data`` に収めていく。 | |||
ツイートの記録を中断するための文を発見するか、記録したツイート数が :const:`~LIMIT_N_TWEETS` 件に達したら終了する。 | |||
""" | |||
soup: Final[BeautifulSoup] = BeautifulSoup(self._page.text, 'html.parser') ##beautifulsoupでレスポンス解析 | soup: Final[BeautifulSoup] = BeautifulSoup(self._page.text, 'html.parser') ##beautifulsoupでレスポンス解析 | ||
tweets: Final[list[bs4.element.Tag]] = self._get_timeline_items(soup) ##一ツイートのブロックごとにリストで取得 | tweets: Final[list[bs4.element.Tag]] = self._get_timeline_items(soup) ##一ツイートのブロックごとにリストで取得 | ||
530行目: | 703行目: | ||
self._make_txt() | self._make_txt() | ||
def _escape_wiki_reserved_words(self, text: str) -> str: | def _escape_wiki_reserved_words(self, text: str) -> str: | ||
"""MediaWikiの文法と衝突する文字を無効化する。 | |||
Args: | |||
text str: ツイートの文字列。 | |||
Returns: | |||
str: MediaWikiの文法と衝突する文字がエスケープされたツイートの文字列。 | |||
""" | |||
def escape_nolink_urls(text: str) -> str: | def escape_nolink_urls(text: str) -> str: | ||
"""Archiveテンプレートの中にないURLがWikiでaタグに変換されないよう無効化する。 | |||
Args: | |||
text str: ツイートの文字列。 | |||
Returns: | |||
str: Archiveテンプレートの中にないURLがnowikiタグでエスケープされた文字列。 | |||
""" | |||
is_in_archive_template: bool = False | is_in_archive_template: bool = False | ||
i: int = 0 | i: int = 0 | ||
560行目: | 748行目: | ||
return text | return text | ||
def _archive_soup(self, tag: bs4.element.Tag) -> None: | def _archive_soup(self, tag: bs4.element.Tag) -> None: | ||
"""ツイート内のaタグをテンプレートArchiveの文字列に変化させる。 | |||
NitterリンクをYouTubeへのリンクに、bibliogramへのリンクをInstagramへのリンクに修正する。 | |||
Args: | |||
tag bs4.element.Tag: ツイート内容の本文である ``.media-body`` タグを表すbeautifulsoup4タグ。 | |||
""" | |||
urls_in_tweet: Final[bs4.element.ResultSet] = tag.find_all('a') | urls_in_tweet: Final[bs4.element.ResultSet] = tag.find_all('a') | ||
for url in urls_in_tweet: | for url in urls_in_tweet: | ||
595行目: | 789行目: | ||
url.replace_with(self._archive_url(url_link, url_text)) ##テンプレートArchiveに変化 | url.replace_with(self._archive_url(url_link, url_text)) ##テンプレートArchiveに変化 | ||
def _archive_url(self, url: Final[str], text: Final[str|None] = None) -> str: | def _archive_url(self, url: Final[str], text: Final[str|None] = None) -> str: | ||
if '#' in url: | """URLをArchiveテンプレートでラップする。 | ||
フラグメント識別子がURLに含まれていたら、Archive側のURLにも反映させる。 | |||
Args: | |||
url Final[str]: ラップするURL。 | |||
text Final[str|None]: ArchiveテンプレートでURLの代わりに表示する文字列。 | |||
Returns: | |||
str: ArchiveタグでラップしたURL。 | |||
""" | |||
if '#' in url: # フラグメント識別子の処理 | |||
main_url, fragment = url.split('#', maxsplit=1) | main_url, fragment = url.split('#', maxsplit=1) | ||
if text is None: | if text is None: | ||
609行目: | 813行目: | ||
return '{{Archive|1=' + unquote(url) + '|2=' + self._archive(url) + '|3=' + text + '}}' ##テンプレートArchiveの文字列返す | return '{{Archive|1=' + unquote(url) + '|2=' + self._archive(url) + '|3=' + text + '}}' ##テンプレートArchiveの文字列返す | ||
def _callinshowlink_url(self, url: Final[str]) -> str: | def _callinshowlink_url(self, url: Final[str]) -> str: | ||
return '{{CallinShowLink|1=' + url + '|2=' + self._archive(url) + '}}' | """URLをCallinShowLinkテンプレートでラップする。 | ||
Args: | |||
url Final[str]: ラップするURL。 | |||
Returns: | |||
str: CallinShowLinkタグでラップしたURL。 | |||
""" | |||
return '{{CallinShowLink|1=' + url + '|2=' + self._archive(url) + '}}' | |||
def _archive(self, url: Final[str]) -> str: | def _archive(self, url: Final[str]) -> str: | ||
"""URLから対応するarchive.todayのURLを返す。 | |||
取得できれば魚拓ページのURLを返す。 | |||
魚拓がなければその旨の警告を表示し、https://archive.ph/https%3A%2F%2Fxxxxxxxxの形で返される。 | |||
アクセスに失敗すればその旨の警告を表示し、https://archive.ph/https%3A%2F%2Fxxxxxxxxの形で返される。 | |||
Args: | |||
url Final[str]: 魚拓を取得するURL。 | |||
Returns: | |||
str: 魚拓のURL。 | |||
""" | |||
archive_url: str = urljoin(self.ARCHIVE_TODAY_STANDARD, quote(unquote(url), safe='&=+?%')) ##wikiに載せるとき用URLで失敗するとこのままhttps://archive.ph/https%3A%2F%2Fxxxxxxxxの形で返される | archive_url: str = urljoin(self.ARCHIVE_TODAY_STANDARD, quote(unquote(url), safe='&=+?%')) ##wikiに載せるとき用URLで失敗するとこのままhttps://archive.ph/https%3A%2F%2Fxxxxxxxxの形で返される | ||
res: Final[Response | None] = self._request(urljoin(self.ARCHIVE_TODAY, quote(unquote(url), safe='&=+?%'))) ##アクセス用URL使って結果を取得 | res: Final[Response | None] = self._request(urljoin(self.ARCHIVE_TODAY, quote(unquote(url), safe='&=+?%'))) ##アクセス用URL使って結果を取得 | ||
628行目: | 850行目: | ||
return archive_url | return archive_url | ||
def go_to_new_page(self) -> None | NoReturn: | def go_to_new_page(self) -> None | NoReturn: | ||
"""Nitterで次のページに移動する。 | |||
次のページが無ければプログラムを終了する。 | |||
""" | |||
soup: Final[BeautifulSoup] = 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") | show_mores: Final[bs4.element.ResultSet] = soup.find_all(class_="show-more") |