利用者:夜泣き/スクリプト
< 利用者:夜泣き
とりあえず取り急ぎ。バグ報告は利用者・トーク:夜泣き
コード
#!/usr/bin/env python3
# © 2022 恒心教 (Koushinism)
# Released under the MIT license
# https://opensource.org/licenses/mit-license.php
"""Twitter自動収集スクリプト
ver4.1.8 2023/11/26恒心
当コードは恒心停止してしまった https://rentry.co/7298g の降臨ショーツイート自動収集スクリプトの復刻改善版です
前開発者との出会いに感謝
Examples:
定数類は状況に応じて変えてください。:class:`~UserProperties` で変更できます。
::
$ python3 (ファイル名)
オプションに ``--krsw`` とつけると自動モードになります。
::
$ python3 (ファイル名) --krsw
自動モードではユーザーは降臨ショー、クエリはなし、取得上限まで自動です。
つまりユーザー入力が要りません。
``--no-browser`` オプションでTor Browserを使用しないモードに、
``--disable-script`` オプションでTor BrowserのJavaScriptを使用しないモードになります。
``--search-unarchived`` オプションでは、`archive.today <https://archive.today>`_ から
Wikiに未掲載のツイートのURLを収集するモードになります。
Note:
* Pythonのバージョンは3.12以上
* 環境は玉葱前提です。
* TailsやWhonixでない場合、Tor Browserを入れておくか、torコマンドでプロキシを立てておくことが必要です。
* Whonix-Workstation, MacOSで動作確認済
* MacOSの場合はTor Browserをダウンロードするかbrewでtorコマンドを導入してから実行する
* PySocks, bs4, seleniumはインストールしないと標準で入ってません
* requestsも環境によっては入っていないかもしれない
* ``$ python3 -m pip install bs4 requests PySocks selenium``
* pipも入っていなければ ``$ sudo apt install pip``
* `ffmpeg <https://ffmpeg.org>`_ が入っていると動画も自動取得しますが、無くても動きます
* バグ報告は `利用者・トーク:夜泣き#利用者:夜泣き/スクリプトについて \
<https://krsw-wiki.org/wiki/利用者・トーク:夜泣き#利用者:夜泣き/スクリプトについて>`_
"""
import codecs
import json
import logging
import os
import platform
import random
import re
import shutil
import subprocess
import sys
from abc import ABCMeta, abstractmethod
from argparse import ArgumentParser, Namespace
from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from re import Match, Pattern
from time import sleep
from types import MappingProxyType, TracebackType
from typing import (Final, NamedTuple, NoReturn, Self, assert_never, final,
override)
from urllib.parse import (ParseResult, parse_qs, quote, unquote, urljoin,
urlparse)
from zoneinfo import ZoneInfo
import requests
from bs4 import BeautifulSoup
from bs4.element import NavigableString, ResultSet, Tag
from selenium import webdriver
from selenium.common.exceptions import (InvalidSwitchToTargetException,
WebDriverException)
from selenium.webdriver.common.by import By
from selenium.webdriver.firefox.options import Options as FirefoxOptions
from selenium.webdriver.support import expected_conditions as ec
from selenium.webdriver.support.wait import WebDriverWait
logging.basicConfig(format='{asctime} [{levelname:.4}] : {message}', style='{')
logger: Final[logging.Logger] = logging.getLogger(__name__)
logger.setLevel(logging.INFO) # basicConfigで設定するとモジュールのDEBUGログなども出力される
@dataclass(init=False, eq=False, frozen=True)
class UserProperties:
"""ユーザ設定。
実行時に変更する可能性のある定数はここで定義する。
"""
media_dir: Final[str] = 'tweet_media'
"""Final[str]: 画像などのツイートメディアをダウンロードするディレクトリ。
"""
filename: Final[str] = 'tweet.txt'
"""Final[str]: ツイートを保存するファイルの名前。
"""
@dataclass(init=False, eq=False, frozen=True)
class NitterCrawler:
"""``--search-unarchived`` オプション無しの時に利用する設定値。
"""
limit_n_tweets: Final[int] = 100
"""Final[int]: 取得するツイート数の上限。
"""
report_interval: Final[int] = 5
"""Final[int]: 記録件数を報告するインターバル。
"""
nitter_instance: Final[str] = 'http://nitter.pjsfkvpxlinjamtawaksbnnaqs2fc2mtvmozrzckxh7f3kis6yea25ad.onion/'
"""Final[str]: Nitterのインスタンス。
生きているのは https://github.com/zedeus/nitter/wiki/Instances で確認。
Note:
末尾にスラッシュ必須。
インスタンスによっては画像の取得ができない。
Ｔor用のインスタンスでないと動画の取得ができない。
"""
@dataclass(init=False, eq=False, frozen=True)
class ArchiveCrawler:
"""``--search-unarchived`` オプションを付けたときに使用する設定値。
"""
tweet_url_prefix_default: Final[str] = '172'
"""Final[str]: ツイートURLの数字部分のうち、予め固定しておく部分。
URLの数字部分がこの数字で始まるもののみをクロールする。
Example:
`tweet_url_prefix_default = 172`、`incremented_num_default = 3` の場合、
`https://twitter.com/CallinShow/status/1723*` から
`https://twitter.com/CallinShow/status/1729*` までを魚拓から検索する。
See Also:
:const:`~incremented_num_default`
"""
incremented_num_default: Final[int] = 0
"""Final[int]: ツイートURLの数字部分うち、インクリメントする桁のデフォルト値。
See Also:
:const:`~tweet_url_prefix_default`
"""
url_list_filename: Final[str] = 'url_list.txt'
"""Final[str]: URLのリストをダンプするファイル名。
"""
class AccessError(Exception):
"""RequestsとSeleniumで共通のアクセスエラー。
"""
pass
class ReCaptchaRequiredError(Exception):
"""JavaScriptがオフの時にreCAPTCHAを要求された場合のエラー。
"""
pass
class AbstractAccessor(metaclass=ABCMeta):
"""HTTPリクエストでWebサイトに接続するための基底クラス。
"""
WAIT_TIME: Final[int] = 1
"""Final[int]: HTTPリクエスト成功失敗関わらず待機時間（秒）。
Note:
1秒待つだけで行儀がいいクローラーだそうなので既定では1秒。
しかし日本のポリホーモは1秒待っていても捕まえてくるので注意 [1]_。
References:
.. [1] 岡崎市立中央図書館事件. (2022, June 21). In Wikipedia.
https://ja.wikipedia.org/wiki/?curid=2187212&oldid=79121945
"""
WAIT_RANGE: Final[int] = 5
"""Final[int]: ランダムな時間待機するときの待機時間の幅（秒）。
"""
REQUEST_TIMEOUT: Final[int] = 30
"""Final[int]: HTTPリクエストのタイムアウト秒数。
"""
@abstractmethod
def get(self, url: str) -> str:
"""URLにアクセスして、HTMLを取得する。
Args:
url (str): 接続するURL。
Raises:
AccessError: 通信に失敗した場合のエラー。
Returns:
str: レスポンスのHTML。
"""
...
@abstractmethod
def set_cookies(self, cookies: dict[str, str]) -> None:
"""Cookieをセットする。
Args:
cookies (dict[str, str]): Cookieのキーバリューペア。
"""
...
@final
def _random_sleep(self) -> None:
"""ランダムな秒数スリープする。
自動操縦だとWebサイトに見破られないため。
"""
sleep(random.randrange(self.WAIT_TIME,
self.WAIT_TIME + self.WAIT_RANGE))
class RequestsAccessor(AbstractAccessor):
"""RequestsモジュールでWebサイトに接続するためのクラス。
"""
HEADERS: Final[dict[str, str]] = {
'User-Agent':
'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/118.0'
}
"""Final[dict[str, str]]: HTTPリクエスト時のヘッダ。
"""
PROXIES_WITH_BROWSER: Final[dict[str, str]] = {
'http': 'socks5h://127.0.0.1:9150',
'https': 'socks5h://127.0.0.1:9150'
}
"""Final[dict[str, str]]: Tor Browserを起動しているときのHTTPプロキシの設定。
"""
PROXIES_WITH_COMMAND: Final[dict[str, str]] = {
'http': 'socks5h://127.0.0.1:9050',
'https': 'socks5h://127.0.0.1:9050'
}
"""Final[dict[str, str]]: torコマンドを起動しているときのHTTPプロキシの設定。
"""
TOR_CHECK_URL: Final[str] = 'https://check.torproject.org/api/ip'
"""Final[str]: Ｔor経由で通信しているかチェックするサイトのURL。
"""
def __init__(self) -> None:
"""コンストラクタ。
"""
# Ｔorに必要なプロキシをセット
self._cookies: dict[str, str] = {}
self._proxies: dict[str, str] | None = self._choose_tor_proxies()
def _execute(self, url: str) -> requests.models.Response:
"""引数のURLにRequestsモジュールでHTTP接続する。
Args:
url (str): 接続するURL。
Raises:
requests.HTTPError: ステータスコードが200でない場合のエラー。
Returns:
requests.models.Response: レスポンスのオブジェクト。
"""
sleep(self.WAIT_TIME) # DoS対策で待つ
res: Final[requests.models.Response] = requests.get(
url,
timeout=self.REQUEST_TIMEOUT,
headers=self.HEADERS,
allow_redirects=False,
proxies=self._proxies,
cookies=self._cookies)
res.raise_for_status() # HTTPステータスコードが200番台以外でエラー発生
return res
@override
def get(self, url: str) -> str:
try:
return self._execute(url).text
except (requests.HTTPError, requests.ConnectionError) as e:
raise AccessError from e
@override
def set_cookies(self, cookies: dict[str, str]) -> None:
self._cookies.update(cookies)
def get_image(self, url: str) -> bytes | None:
"""引数のURLから画像のバイナリ列を取得する。
Args:
url (str): 接続するURL。
Raises:
AccessError: アクセスに失敗した場合のエラー。
Returns:
bytes | None: 画像のバイナリ。画像でなければ `None`。
"""
try:
res: Final[requests.models.Response] = self._execute(url)
except (requests.HTTPError, requests.ConnectionError) as e:
raise AccessError from e
if 'image' in res.headers['content-type']:
return res.content
else:
return None
def _choose_tor_proxies(self) -> dict[str, str] | None | NoReturn:
"""Ｔorを使うのに必要なプロキシ情報を返す。
プロキシなしで接続できれば `None`、
Tor Browserのプロキシで接続できるなら :const:`~PROXIES_WITH_BROWSER`、
torコマンドのプロキシで接続できるなら :const:`~PROXIES_WITH_COMMAND`。
いずれでもアクセスできなければ異常終了する。
Raises:
AccessError: :const:`~TOR_CHECK_URL` にアクセスできない場合のエラー。
Returns:
dict[str, str] | None | NoReturn: プロキシ情報。
"""
logger.info('Ｔorのチェック中ですを')
# プロキシなしでTorにアクセスできるかどうか
self._proxies = None
res: str = self._execute(self.TOR_CHECK_URL).text
is_tor: bool = json.loads(res)['IsTor']
if is_tor:
logger.info('Ｔor connection OK')
return None
# Tor BrowserのプロキシでTorにアクセスできるかどうか
try:
self._proxies = self.PROXIES_WITH_BROWSER
res = self._execute(self.TOR_CHECK_URL).text
is_tor = json.loads(res)['IsTor']
if is_tor:
logger.info('Ｔor browser connection OK')
return self.PROXIES_WITH_BROWSER
except requests.exceptions.ConnectionError:
pass
# torコマンドのプロキシでTorにアクセスできるかどうか
try:
self._proxies = self.PROXIES_WITH_COMMAND
res = self._execute(self.TOR_CHECK_URL).text
is_tor = json.loads(res)['IsTor']
if is_tor:
logger.info('Ｔor proxy OK')
return self.PROXIES_WITH_COMMAND
else:
raise AccessError('サイトにＴorのIPでアクセスできていないなりを')
except requests.exceptions.ConnectionError as e:
logger.critical(e)
logger.critical('通信がＴorのSOCKS proxyを経由していないなりを')
sys.exit(1)
@property
def proxies(self) -> dict[str, str] | None:
"""オブジェクトのプロキシ設定を返す。
Returns:
dict[str, str] | None: プロキシ設定。
"""
return self._proxies
class SeleniumAccessor(AbstractAccessor):
"""SeleniumでWebサイトに接続するためのクラス。
"""
TOR_BROWSER_PATHS: Final[MappingProxyType[str, str]] = MappingProxyType({
'Windows': r'C:\Program Files\Tor Browser\Browser\firefox.exe',
'Darwin': '/Applications/Tor Browser.app/Contents/MacOS/firefox',
'Linux': '/usr/bin/torbrowser'
})
"""Final[MappingProxyType[str, str]]: OSごとのTor Browserのパス。
"""
WEB_DRIVER_WAIT_TIME: Final[int] = 15
"""Final[int]: 最初のTor接続時の待機時間（秒）。
"""
WAIT_TIME_FOR_RECAPTCHA: Final[int] = 10_000
"""Final[int]: reCAPTCHAのための待機時間（秒）。
"""
def __init__(self, enable_javascript: bool) -> None:
"""コンストラクタ。
Tor Browserを自動操縦するためのSeleniumドライバを初期化する。
Args:
enable_javascript (bool): JavaScriptを有効にするかどうか。reCAPTCHA対策に必要。
"""
self._options: Final[FirefoxOptions] = FirefoxOptions()
self._options.binary_location = (
self.TOR_BROWSER_PATHS[platform.system()]
)
self._options.add_argument('--user-data-dir=selenium') # type: ignore
if enable_javascript:
logger.warning('reCAPTCHA対策のためJavaScriptをonにしますを')
self._options.preferences.update({ # type: ignore
'javascript.enabled': enable_javascript,
'intl.accept_languages': 'en-US, en',
'intl.locale.requested': 'US',
'font.language.group': 'x-western',
'dom.webdriver.enabled': False # 自動操縦と見破られないための設定
})
self._refresh_browser()
def quit(self) -> None:
"""Seleniumドライバを終了する。
"""
if hasattr(self, '_driver'):
logger.debug('ブラウザ終了')
self._driver.quit()
def _refresh_browser(self) -> None:
"""ブラウザを起動する。
"""
try:
logger.debug('ブラウザ起動')
self._driver: webdriver.Firefox = webdriver.Firefox(
options=self._options)
sleep(1)
web_driver_wait: Final[WebDriverWait] = WebDriverWait(
self._driver,
self.WEB_DRIVER_WAIT_TIME)
web_driver_wait.until(
ec.element_to_be_clickable((By.ID, 'connectButton'))
)
self._driver.find_element(By.ID, 'connectButton').click()
# Torの接続が完了するまで待つ
web_driver_wait.until(ec.url_contains('about:blank'))
except BaseException:
self.quit()
raise
def _check_recaptcha(self, url: str) -> None:
"""reCAPTCHAが表示されているかどうか判定して、入力を待機する。
botであることが検知された場合、自動でブラウザを再起動する。
Args:
url (str): アクセスしようとしているURL。
reCAPTCHAが要求されると `current_url` が変わることがあるので必要。
Raises:
ReCaptchaRequiredError: JavaScriptがオフの状態でreCAPTCHAが要求された場合のエラー。
"""
iframe_selector: Final[str] = (
'iframe[title="recaptcha challenge expires in two minutes"]')
if len(self._driver.find_elements(By.ID, 'g-recaptcha')) > 0:
if self._options.preferences.get('javascript.enabled'): # type: ignore
logger.warning(f'{url} でreCAPTCHAが要求されたナリ')
print('reCAPTCHAを解いてね（笑）、それはできるよね。')
print('botバレしたら自動でブラウザが再起動するナリよ')
WebDriverWait(self._driver, self.WEB_DRIVER_WAIT_TIME).until(
ec.presence_of_element_located(
(By.CSS_SELECTOR, iframe_selector)
)
)
self._driver.switch_to.frame( # reCAPTCHAのフレームに遷移する
self._driver.find_element(By.CSS_SELECTOR, iframe_selector)
)
try:
WebDriverWait(
self._driver, self.WAIT_TIME_FOR_RECAPTCHA
).until(
ec.visibility_of_element_located(
# bot検知された場合に現れるクラス
(By.CLASS_NAME, 'rc-doscaptcha-header')
)
)
except InvalidSwitchToTargetException:
# reCAPTCHAのフレームがなくなっていた場合
logger.info('reCAPTCHAが解かれましたを')
self._driver.switch_to.default_content()
self._random_sleep() # DoS対策で待つ
else:
# waitを普通に抜けた場合
logger.warning('botバレしたなりを')
# 一回ブラウザを落として起動し直す
self.quit()
self._refresh_browser()
self.get(url)
else:
raise ReCaptchaRequiredError(
f'JavaScriptがオフの状態でreCAPTCHAが要求されました。 URL: {url}')
@override
def get(self, url: str) -> str:
self._random_sleep() # DoS対策で待つ
try:
self._driver.get(url)
WebDriverWait(self._driver, self.WEB_DRIVER_WAIT_TIME).until_not(
ec.title_is('Redirecting')) # Nitterのリダイレクトを検知する
self._check_recaptcha(url)
except WebDriverException as e:
# Selenium固有の例外を共通の例外に変換
raise AccessError from e
return self._driver.page_source
@override
def set_cookies(self, cookies: dict[str, str]) -> None:
for name, value in cookies.items():
self._driver.add_cookie( # type: ignore
{'name': name, 'value': value})
self._driver.refresh()
class AccessorHandler:
"""WebサイトからHTMLを取得するためのクラス。
RequestsとSeleniumのどちらかを選択して使用することができ、その違いを隠蔽する。
"""
LIMIT_N_REQUESTS: Final[int] = 5
"""Final[int]: HTTPリクエスト失敗時の再試行回数。
"""
WAIT_TIME_FOR_ERROR: Final[int] = 4
"""Final[int]: HTTPリクエスト失敗時にさらに追加する待機時間。
"""
def __init__(self, use_browser: bool, enable_javascript: bool) -> None:
"""コンストラクタ。
Requestのみを利用するか、Seleniumも利用するか引数で選択して初期化する。
Args:
use_browser (bool): `True` ならSeleniumを利用する。
`False` ならRequestsのみでアクセスする。
enable_javascript (bool): SeleniumでJavaScriptを利用する場合は `True`。
"""
self._selenium_accessor: Final[SeleniumAccessor | None] = (
SeleniumAccessor(enable_javascript) if use_browser else None
)
self._requests_accessor: Final[RequestsAccessor] = RequestsAccessor()
self._last_url = '' # 最後にアクセスしたURL
def __enter__(self) -> Self:
"""`with` ブロックの開始時に実行する。
Returns:
Self: オブジェクト自身。
"""
return self
def __exit__(self,
exc_type: type[BaseException] | None,
exc_value: BaseException | None,
traceback: TracebackType | None) -> None:
"""`with` ブロックの終了時に実行する。
Args:
exc_type (type[BaseException] | None): コンテキスト内で例外を吐いた場合の例外タイプ。
exc_value (BaseException | None): コンテキスト内で例外を吐いた場合の例外。
traceback (TracebackType | None): コンテキスト内で例外を吐いた場合のトレースバック。
"""
if self._selenium_accessor is not None:
self._selenium_accessor.quit()
def request_once(self, url: str) -> str:
"""引数のURLにHTTP接続する。
Args:
url (str): 接続するURL。
Returns:
str: レスポンスのテキスト。
Raises:
AccessError: アクセスエラー。
Note:
失敗かどうかは呼出側で要判定。
"""
try:
if self._selenium_accessor is not None:
return self._selenium_accessor.get(url)
else:
return self._requests_accessor.get(url)
except AccessError:
raise
def _request_with_callable[T](
self,
url: str,
request_callable: Callable[[str], T]) -> T | None:
"""`request_callable` の実行を :const:`~LIMIT_N_REQUESTS` で定義された回数まで試す。
成功すると結果を返す。
接続失敗が何度も起きると `None` を返す。
Args:
url (str): 接続するURL
request_callable (Callable[[str], T]): 1回リクエストを行うメソッド。
Returns:
T | None: レスポンス。接続失敗が何度も起きると `None` を返す。
"""
assert url, 'URLが空っぽでふ'
logger.debug('Requesting ' + unquote(url))
self._last_url = url
for i in range(self.LIMIT_N_REQUESTS):
try:
res: Final[T] = request_callable(url)
except AccessError:
logger.warning(
url + 'への通信失敗ナリ '
f'{i + 1}/{self.LIMIT_N_REQUESTS}回')
if i < self.LIMIT_N_REQUESTS:
sleep(self.WAIT_TIME_FOR_ERROR) # 失敗時は長めに待つ
else:
return res
return None
def request(self, url: str) -> str | None:
"""HTTP接続を :const:`~LIMIT_N_REQUESTS` で定義された回数まで試す。
成功すると結果を返す。
接続失敗が何度も起きると `None` を返す。
Args:
url (str): 接続するURL。
Returns:
str | None: レスポンスのテキスト。接続失敗が何度も起きると `None` を返す。
Note:
失敗かどうかは呼出側で要判定
"""
return self._request_with_callable(url, self.request_once)
def request_with_requests_module(self, url: str) -> str | None:
"""RequestsモジュールでのHTTP接続を :const:`~LIMIT_N_REQUESTS` で定義された回数まで試す。
成功すると結果を返す。
接続失敗が何度も起きると `None` を返す。
Args:
url (str): 接続するURL。
Returns:
str | None: レスポンスのテキスト。接続失敗が何度も起きると `None` を返す。
Note:
失敗かどうかは呼出側で要判定
"""
return self._request_with_callable(url, self._requests_accessor.get)
def request_image(self, url: str) -> bytes | None:
"""Requestsモジュールで画像ファイルを取得する。
成功すると結果を返す。
接続失敗が何度も起きると `None` を返す。
Args:
url (str): 接続するURL。
Returns:
bytes | None: レスポンスのバイト列。接続失敗が何度も起きると `None` を返します。
Note:
失敗かどうかは呼出側で要判定
"""
return self._request_with_callable(
url,
self._requests_accessor.get_image)
def set_cookies(self, cookies: dict[str, str]) -> None:
"""Cookieをセットする。
Args:
cookies (dict[str, str]): Cookieのキーバリューペア。
"""
self._requests_accessor.set_cookies(cookies)
if self._selenium_accessor is not None:
self._selenium_accessor.set_cookies(cookies)
@property
def last_url(self) -> str:
"""最後にアクセスしたURLを返す。
Returns:
str: 最後にアクセスしたURL。
"""
return self._last_url
@property
def proxies(self) -> dict[str, str] | None:
"""RequestsAccessorオブジェクトのプロキシ設定を返す。
Returns:
dict[str, str] | None: RequestsAccessorオブジェクトのプロキシ設定。
"""
return self._requests_accessor.proxies
class TableBuilder:
"""Wikiの表を組み立てるためのクラス。
"""
FILENAME: Final[str] = UserProperties.filename
"""Final[str]: ツイートを保存するファイルの名前。
"""
def __init__(self, date: datetime | None = None) -> None:
"""コンストラクタ。
Args:
date (datetime | None, optional): 記録するツイートの最新日付。デフォルトは今日の日付。
"""
self._tables: Final[list[str]] = ['']
self._count: int = 0 # 記録数
self._date: datetime = date or datetime.today()
@property
def count(self) -> int:
"""表に追加したツイートの件数を返す。
Returns:
int: 表に追加したツイートの件数。
"""
return self._count
def append(self, callinshow_template: str, text: str) -> None:
"""ツイートを表に追加する。
Args:
callinshow_template (str): ツイートのURLをCallinshowLinkテンプレートに入れたもの。
text (str): ツイートの本文。
"""
self._tables[-1] = (
'!' + callinshow_template + '\n|-\n|\n'
+ text
+ '\n|-\n'
+ self._tables[-1])
self._count += 1
def dump_file(self) -> None:
"""Wikiテーブルをファイル出力する。
"""
self._next_day()
result_txt: Final[str] = '\n'.join(reversed(self._tables))
with codecs.open(self.FILENAME, 'w', 'utf-8') as f:
f.write(result_txt)
logger.info('テキストファイル手に入ったやで〜')
def next_day_if_necessary(self, date: datetime) -> None:
"""引数dateがインスタンスの持っている日付より前の場合、日付更新処理をする。
Args:
date (datetime): 次のツイートの日付。
"""
if (date.year != self._date.year or date.month != self._date.month
or date.day != self._date.day):
self._next_day(date)
def _next_day(self, date: datetime | None = None) -> None:
"""Wikiテーブルに日付の見出しを付与し、日付を更新する。
記録済みのツイートが1つもない場合は日付の更新だけをする。
Args:
date (datetime | None, optional): 次に記録するツイートの日付。
"""
if self._tables[-1]:
self._tables[-1] = self._convert_to_text_table(self._tables[-1])
if platform.system() == 'Windows':
self._tables[-1] = self._date.strftime(
'\n=== %#m月%#d日 ===\n') + self._tables[-1]
logger.info(self._date.strftime('%#m月%#d日のツイートを取得完了ですを'))
else:
self._tables[-1] = self._date.strftime(
'\n=== %-m月%-d日 ===\n') + self._tables[-1]
logger.info(self._date.strftime('%-m月%-d日のツイートを取得完了ですを'))
if date is not None:
self._date = date
if self._tables[-1]:
self._tables.append('')
def _convert_to_text_table(self, text: str) -> str:
"""Wikiでテーブル表示にするためのヘッダとフッタをつける。
Args:
text (str): ヘッダとフッタがないWikiテーブル。
Returns:
str: テーブル表示用のヘッダとフッタがついたWikiテーブル。
"""
return '{|class="wikitable" style="text-align: left;"\n' + text + '|}'
@classmethod
def escape_wiki_reserved_words(cls, text: str) -> str:
"""MediaWikiの文法と衝突する文字を無効化する。
Args:
text (str): ツイートの文字列。
Returns:
str: MediaWikiの文法と衝突する文字がエスケープされたツイートの文字列。
"""
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
i: int = 0
while i < len(text):
if is_in_archive_template:
if text[i:i + len('}}')] == '}}':
is_in_archive_template = False
i += len('}}')
else:
if (text[i:i + len('{{Archive|')] == '{{Archive|'
or text[i:i + len('{{Archive|')] == '{{archive|'):
is_in_archive_template = True
i += len('{{Archive|')
elif text[i:i + len('https://')] == 'https://':
text = (text[:i]
+ '<nowiki>https://</nowiki>'
+ text[i + len('https://'):])
i += len('<nowiki>https://</nowiki>')
elif text[i:i + len('http://')] == 'http://':
text = (text[:i]
+ '<nowiki>http://</nowiki>'
+ text[i + len('http://'):])
i += len('<nowiki>http://</nowiki>')
i += 1
return text
if not hasattr(cls, '_escape_callables'):
# 初回呼び出しの時だけ正規表現をコンパイルする
head_space_pattern: Final[Pattern[str]] = re.compile(
r'^ ', re.MULTILINE)
head_marks_pattern: Final[Pattern[str]] = re.compile(
r'^([\*#:;])', re.MULTILINE)
bar_pattern: Final[Pattern[str]] = re.compile(
r'^----', re.MULTILINE)
cls._escape_callables: tuple[Callable[[str], str], ...] = (
lambda t: t.replace('\n', '<br>\n'),
lambda t: head_space_pattern.sub(' ', t),
lambda t: head_marks_pattern.sub(r'<nowiki>\1</nowiki>', t),
lambda t: bar_pattern.sub('<nowiki>----</nowiki>', t),
lambda t: escape_nolink_urls(t),
)
escaped_text: str = text
for escape_callable in cls._escape_callables:
escaped_text = escape_callable(escaped_text)
return escaped_text
@staticmethod
def archive_url(
url: str,
archived_url: str,
text: str | None = None) -> str:
"""URLをArchiveテンプレートでラップする。
Args:
url (str): ラップするURL。
archive_url (str): ラップするURLの魚拓のURL。
text (str | None, optional): ArchiveテンプレートでURLの代わりに表示する文字列。
Returns:
str: ArchiveタグでラップしたURL。
"""
if text is None:
return '{{Archive|1=' + unquote(url) + '|2=' + archived_url + '}}'
else:
return '{{Archive|1=' + unquote(url) + '|2=' + archived_url \
+ '|3=' + text + '}}'
@staticmethod
def callinshowlink_url(url: str, archived_url: str) -> str:
"""URLをCallinShowLinkテンプレートでラップする。
Args:
url (str): ラップするURL。
archive_url (str): ラップするURLの魚拓のURL。
Returns:
str: CallinShowLinkタグでラップしたURL。
"""
return '{{CallinShowLink|1=' + url + '|2=' + archived_url + '}}'
class FfmpegStatus(Enum):
"""ffmpegでの動画保存ステータス。
"""
MP4 = 1
"""mp4の取得に成功したときのステータス。
"""
TS = 2
"""tsの取得までは成功したが、mp4への変換に失敗したときのステータス。
"""
FAILED = 3
"""m3u8からtsの取得に失敗したときのステータス。
"""
class TwitterArchiver:
"""ツイートをWikiの形式にダンプするクラス。
Nitterからツイートを取得し、Wikiの形式にダンプする。
削除されたツイートや編集前のツイートは取得できない。
"""
NITTER_INSTANCE: Final[str] = UserProperties.NitterCrawler.nitter_instance
"""Final[str]: Nitterのインスタンス。
生きているのは https://github.com/zedeus/nitter/wiki/Instances で確認。
Note:
末尾にスラッシュ必須。
インスタンスによっては画像の取得ができない。
Ｔor用のインスタンスでないと動画の取得ができない。
"""
ARCHIVE_TODAY: Final[str] = 'http://archiveiya74codqgiixo33q62qlrqtkgmcitqx5u2oeqnmn5bpcbiyd.onion/'
"""Final[str]: archive.todayの魚拓のonionドメイン。
ページにアクセスして魚拓があるか調べるのにはonion版のarchive.todayを使用。
Note:
末尾にスラッシュ必須。
"""
ARCHIVE_TODAY_STANDARD: Final[str] = 'https://archive.vn/'
"""Final[str]: archive.todayの魚拓のクリアネットドメイン。
Wiki上の記事にはクリアネット用のarchive.todayリンクを貼る。
Note:
末尾にスラッシュ必須。
"""
TWITTER_URL: Final[str] = 'https://twitter.com/'
"""Final[str]: TwitterのURL。
Note:
末尾にスラッシュ必須。
"""
TWITTER_MEDIA_URL: Final[str] = 'https://pbs.twimg.com/media/'
"""Final[str]: TwitterのメディアのURL。
"""
INVIDIOUS_INSTANCES_URL: Final[str] = \
'https://api.invidious.io/instances.json'
"""Final[str]: Invidiousのインスタンスのリストを取得するAPIのURL。
"""
INVIDIOUS_INSTANCES_TUPLE: Final[tuple[str, ...]] = (
'piped.kavin.rocks',
'piped.video'
)
"""Final[tuple[str, ...]]: よく使われるInvidiousインスタンスのリスト。
:const:`~INVIDIOUS_INSTANCES_URL` にアクセスしてもインスタンスが取得できないことがあるため、
それによってURLが置換できないことを防ぐ。
"""
CALLINSHOW: Final[str] = 'CallinShow'
"""Final[str]: 降臨ショーのユーザーネーム。
"""
LIMIT_N_TWEETS: Final[int] = UserProperties.NitterCrawler.limit_n_tweets
"""Final[int]: 取得するツイート数の上限。
"""
REPORT_INTERVAL: Final[int] = UserProperties.NitterCrawler.report_interval
"""Final[int]: 記録件数を報告するインターバル。
"""
TWEETS_OR_REPLIES: Final[str] = 'with_replies'
"""Final[str]: NitterのURLのドメインとユーザーネーム部分の接続部品。
"""
NITTER_ERROR_TITLE: Final[str] = 'Error|nitter'
"""Final[str]: Nitterでユーザーがいなかったとき返ってくるページのタイトル。
万が一仕様変更で変わったとき用。
"""
NO_ARCHIVE: Final[str] = 'No results'
"""Final[str]: archive.todayで魚拓がなかったときのレスポンス。
万が一仕様変更で変わったとき用。
"""
NEWEST: Final[str] = 'Load newest'
"""Final[str]: Nitterの前ページ読み込み部分の名前。
万が一仕様変更で変わったとき用。
"""
MEDIA_DIR: Final[str] = UserProperties.media_dir
"""Final[str]: 画像などのツイートメディアをダウンロードするディレクトリ。
"""
def __init__(self) -> None:
"""コンストラクタ。
"""
self._check_slash() # スラッシュが抜けてないかチェック
self._has_ffmpeg: Final[bool] = self._check_ffmpeg() # ffmpegがあるかチェック
self._img_ext_pattern: Final[Pattern[str]] = re.compile(
r'%2F([^%]*\.(?:jpg|jpeg|png|gif))')
self._url_fragment_pattern: Final[Pattern[str]] = re.compile(
r'#[^#]*$')
self._url_query_pattern: Final[Pattern[str]] = re.compile(r'\?.*$')
def _set_queries(self, accessor: AccessorHandler, krsw: bool) -> bool:
"""検索条件を設定する。
:class:`~TwitterArchiver` のメンバをセットし、ツイートを収集するアカウントと
検索クエリ、終わりにするツイートを入力させる。
Args:
accessor (AccessorHandler): アクセスハンドラ
krsw (bool): Trueの場合、名前が :const:`~CALLINSHOW` になり、
クエリと終わりにするツイートが無しになる。
Returns:
bool: 処理成功時は `True`。
"""
# ユーザー名取得
if krsw:
logger.info('名前は自動的に' + self.CALLINSHOW + 'にナリます')
self._name: str = self.CALLINSHOW
else:
name_optional: str | None = self._get_name(accessor)
if name_optional is not None:
self._name: str = name_optional
else:
return False
# 検索クエリとページ取得
self._query_strs: list[str] = []
if krsw:
logger.info('クエリは自動的になしにナリます')
else:
self._get_query()
page_optional: str | None = accessor.request(
urljoin(self.NITTER_INSTANCE, self._name + '/'
+ self.TWEETS_OR_REPLIES))
if page_optional is None:
self._on_fail(accessor)
return False
self._nitter_page: str = page_optional
# Nitter用のCookieをセット
accessor.set_cookies({
'infiniteScroll': '',
'proxyVideos': 'on',
'replaceReddit': '',
'replaceYouTube': '',
})
# 終わりにするツイート取得
if krsw:
logger.info('終わりにするツイートは自動的になしにナリます')
self._stop: str = '' if krsw else self._stop_word()
logger.info(
'ユーザー名: @' + self._name
+ ', クエリ: ["' + '", "'.join(self._query_strs)
+ '"], 終わりにする文言: "' + self._stop
+ '"で検索しまふ'
)
# 日付取得
timeline_item: Final[Tag | NavigableString | None] = BeautifulSoup(
self._nitter_page, 'html.parser'
).find(class_='timeline-item')
assert isinstance(timeline_item, Tag)
date: Final[datetime] = self._tweet_date(timeline_item)
self._table_builder: TableBuilder = TableBuilder(date)
return True
def _check_slash(self) -> None | NoReturn:
"""URLの最後にスラッシュが付いていなければエラーを出す。
Returns:
None | NoReturn: すべてのURLが正しければ `None`。失敗したら例外を出す。
Raises:
RuntimeError: URLの最後にスラッシュがついていない場合に出る。
"""
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の末尾をには/が必須です')
def _check_ffmpeg(self) -> bool:
"""ffmpegがインストールされているかチェックする。
Returns:
bool: ffmpegがインストールされているか。
"""
return shutil.which('ffmpeg') is not None
def _check_nitter_instance(
self, accessor: AccessorHandler) -> None | NoReturn:
"""Nitterのインスタンスが生きているかチェックする。
死んでいたらそこで終了。
接続を一回しか試さない :func:`~_request_once` を使っているのは、
激重インスタンスが指定されたとき試行回数増やして偶然成功してそのまま実行されるのを躱すため。
Args:
accessor (AccessorHandler): アクセスハンドラ。
Returns:
None | NoReturn: Nitterにアクセスできれば `None`。できなければ終了。
"""
logger.info('Nitterのインスタンスチェック中ですを')
try:
accessor.request_once(self.NITTER_INSTANCE)
except AccessError as e:
logger.critical(e)
logger.critical('インスタンスが死んでますを')
sys.exit(1)
logger.info('Nitter OK')
def _check_archive_instance(
self, accessor: AccessorHandler) -> None | NoReturn:
"""archive.todayのTor用インスタンスが生きているかチェックする。
Args:
accessor (AccessorHandler): アクセスハンドラ。
Returns:
None | NoReturn: archive.todayのＴorインスタンスにアクセスできれば `None`。できなければ終了。
"""
logger.info('archive.todayのＴorインスタンスチェック中ですを')
try:
accessor.request_once(self.ARCHIVE_TODAY)
except AccessError as e: # エラー発生時は終了
logger.critical(e)
logger.critical('インスタンスが死んでますを')
sys.exit(1)
logger.info('archive.today OK')
def _invidious_instances(
self,
accessor: AccessorHandler) -> tuple[str, ...] | NoReturn:
"""Invidiousのインスタンスのタプルを取得する。
Args:
accessor (AccessorHandler): アクセスハンドラ。
Returns:
tuple[str, ...] | NoReturn: Invidiousのインスタンスのタプル。インスタンスが死んでいれば終了。
"""
logger.info('Invidiousのインスタンスリストを取得中ですを')
invidious_json: Final[str | None] = (
accessor.request_with_requests_module(self.INVIDIOUS_INSTANCES_URL)
)
if invidious_json is None:
logger.critical('Invidiousが死んでますを')
sys.exit(1)
instance_list: Final[list[str]] = []
for instance_info in json.loads(invidious_json):
instance_list.append(instance_info[0])
# よく使われているものはチェック
for invidious_api in self.INVIDIOUS_INSTANCES_TUPLE:
if invidious_api not in instance_list:
instance_list.append(invidious_api)
logger.debug('Invidiousのインスタンス: [' + ', '.join(instance_list) + ']')
return tuple(instance_list)
def _get_name(self, accessor: AccessorHandler) -> str | None:
"""ツイート収集するユーザー名を標準入力から取得する。
何も入力しないと :const:`~CALLINSHOW` を指定する。
Args:
accessor (AccessorHandler): アクセスハンドラ。
Returns:
str | None: ユーザ名。ユーザページの取得に失敗したら `None`。
"""
while True:
print(
'アカウント名を入れなければない。空白だと自動的に'
+ self.CALLINSHOW
+ 'になりますを')
print('> ', end='')
account_str: Final[str] = input()
# 空欄で降臨ショー
if account_str == '':
return self.CALLINSHOW
else:
res: Final[str | None] = accessor.request(
urljoin(self.NITTER_INSTANCE, account_str))
if res is None: # リクエスト失敗判定
self._on_fail(accessor)
return None
soup: Final[BeautifulSoup] = BeautifulSoup(res, 'html.parser')
if soup.title == self.NITTER_ERROR_TITLE:
print(account_str + 'は実在の人物ではありませんでした')
else:
print('最終的に出会ったのが@' + account_str + 'だった。')
logger.info('@' + account_str + 'をクロールしまふ')
return account_str
def _get_query(self) -> None:
"""検索クエリを標準入力から取得する。
"""
print('検索クエリを入れるナリ。複数ある時は一つの塊ごとに入力するナリ。空欄を入れると検索開始ナリ。')
print('例：「無能」→改行→「脱糞」→改行→「弟殺し」→改行→空欄で改行')
print('> ', end='')
query_input: str = input()
# 空欄が押されるまでユーザー入力受付
while query_input != '':
self._query_strs.append(query_input)
print('> ', end='')
query_input = input()
print('クエリのピースが埋まっていく。')
def _on_fail(self, accessor: AccessorHandler) -> None:
"""接続失敗時処理。
取得に成功した分だけファイルにダンプする。
ログにトレースバックも表示する。
Args:
accessor (AccessorHandler): アクセスハンドラ。
"""
logger.critical('接続失敗時処理をしておりまふ', stack_info=True)
logger.critical('最後にアクセスしたURL: ' + accessor.last_url)
print('接続失敗しすぎで強制終了ナリ')
if self._table_builder.count > 0: # 取得成功したデータがあれば発行
print('取得成功した分だけ発行しますを')
self._table_builder.dump_file()
def _stop_word(self) -> str:
"""ツイートの記録を中断するための文をユーザに入力させる。
Returns:
str: ツイートの記録を中断するための文。ツイート内容の一文がその文と一致したら記録を中断する。
"""
print(
'ここにツイートの本文を入れる！すると・・・・なんとそれを含むツイートで記録を中断する！（これは本当の仕様です。）'
f'ちなみに、空欄だと{self.LIMIT_N_TWEETS}件まで終了しない。')
print('> ', end='')
end_str: Final[str] = input()
return end_str
def _download_media(
self,
media_url: str,
media_name: str,
accessor: AccessorHandler) -> bool:
"""ツイートの画像をダウンロードし、:const:`~MEDIA_DIR` に保存する。
Args:
media_url (str): 画像のURL。
media_name (str): 画像ファイル名。Nitter上のimgタグのsrc属性では、
``/pic/media%2F`` に後続する。
accessor (AccessorHandler): アクセスハンドラ。
Returns:
bool: 保存に成功したかどうか。
"""
os.makedirs(self.MEDIA_DIR, exist_ok=True)
image_bytes: Final[bytes | None] = accessor.request_image(media_url)
if image_bytes is not None:
with open(os.path.join(self.MEDIA_DIR, media_name), 'wb') as f:
f.write(image_bytes)
return True
else:
return False
def _download_m3u8(
self,
media_path: str,
ts_filename: str,
mp4_filename: str,
proxies: dict[str, str] | None) -> FfmpegStatus:
"""ffmpegで動画をダウンロードし、:const:`~MEDIA_DIR` に保存する。
Args:
media_path (str): 動画のm3u8ファイルのNitter上でのパス。
ts_filename (str): m3u8から取得したtsファイルのパス。
mp4_filename (str): tsファイルから変換したmp4ファイルのパス。
proxies (dict[str, str] | None): m3u8での通信に用いるプロキシ設定。
Returns:
FfmpegStatus: ffmpegでの保存ステータス。
"""
returncode: Final[int] = subprocess.run(
[
'ffmpeg', '-y',
'-http_proxy', 'proxies["http"]',
'-i', urljoin(self.NITTER_INSTANCE, media_path),
'-c', 'copy', ts_filename
] if proxies is not None else [
'ffmpeg', '-y',
'-i', urljoin(self.NITTER_INSTANCE, media_path),
'-c', 'copy', ts_filename
],
stdout=subprocess.DEVNULL).returncode
# 取得成功したらtsをmp4に変換
if returncode == 0:
ts2mp4_returncode: Final[int] = subprocess.run(
[
'ffmpeg', '-y', '-i', ts_filename,
'-acodec', 'copy', '-vcodec', 'copy', mp4_filename
],
stdout=subprocess.DEVNULL
).returncode
if ts2mp4_returncode == 0:
return FfmpegStatus.MP4
else:
return FfmpegStatus.TS
else:
return FfmpegStatus.FAILED
def _tweet_date(self, tweet: Tag) -> datetime:
"""ツイートの時刻を取得する。
Args:
tweet (Tag): ツイート内容である ``.timeline-item`` タグを表すbeautifulsoup4タグ。
Returns:
datetime: ツイートの時刻。
"""
tweet_date: Final[Tag | NavigableString | None] = tweet.find(
class_='tweet-date')
assert isinstance(tweet_date, Tag)
tweet_date_a: Final[Tag | None] = tweet_date.a
assert tweet_date_a is not None
date_str: Final[str | list[str] | None] = tweet_date_a.get('title')
assert isinstance(date_str, str)
return datetime.strptime(
date_str,
'%b %d, %Y · %I:%M %p %Z').replace(
tzinfo=ZoneInfo('UTC')).astimezone(
ZoneInfo('Asia/Tokyo'))
def _fetch_tweet_media(
self,
tweet: Tag,
tweet_url: str,
accessor: AccessorHandler) -> str:
"""ツイートの画像や動画を取得する。
Args:
tweet (Tag): ツイート内容である ``.timeline-item`` タグを表すbeautifulsoup4タグ。
tweet_url (str): ツイートのURL。
accessor (AccessorHandler): アクセスハンドラ。
Returns:
str: Wiki記法でのファイルへのリンクの文字列。
"""
# 引用リツイート内のメディアを選択しないように.tweet-body直下の.attachmentsを選択
tweet_media: Final[Tag | None] = tweet.select_one(
'.tweet-body > .attachments')
media_txt: str = ''
if tweet_media is not None:
media_list: Final[list[str]] = []
# ツイートの画像の取得
for image_a in tweet_media.select('.attachment.image a'):
try:
href: Final[str | list[str] | None] = image_a.get('href')
assert isinstance(href, str)
img_matched: Final[Match[str] | None] = (
self._img_ext_pattern.search(href))
assert img_matched is not None
media_name: Final[str] = img_matched.group(1)
media_list.append(f'[[ファイル:{media_name}|240px]]')
if self._download_media(
urljoin(self.TWITTER_MEDIA_URL, media_name),
media_name,
accessor):
logger.info(
os.path.join(self.MEDIA_DIR, media_name)
+ ' をアップロードしなければない。')
else:
logger.info(
urljoin(self.TWITTER_MEDIA_URL, media_name)
+ ' をアップロードしなければない。')
except AttributeError:
logger.error(f'{tweet_url}の画像が取得できませんでしたを 当職無能')
media_list.append('[[ファイル:(画像の取得ができませんでした)|240px]]')
# ツイートの動画の取得。ffmpegが入っていなければ手動で取得すること
for i, video_container in enumerate(
tweet_media.select('.attachment.video-container')):
if not self._has_ffmpeg:
logger.warning(f'ffmpegがないため{tweet_url}の動画が取得できませんでしたを')
media_list.append('[[ファイル:(動画の取得ができませんでした)|240px]]')
continue
# videoタグがない場合は取得できない
video: Final[Tag | None] = video_container.select_one('video')
if video is None:
logger.error(f'{tweet_url}の動画が取得できませんでしたを 当職無能')
media_list.append('[[ファイル:(動画の取得ができませんでした)|240px]]')
continue
data_url: Final[str | list[str] | None] = video.get('data-url')
assert isinstance(data_url, str)
video_matched: Final[Match[str] | None] = re.search(
r'[^/]+$', data_url)
assert video_matched is not None
media_path: Final[str] = unquote(video_matched.group())
tweet_id: Final[str] = tweet_url.split('/')[-1]
ts_filename: Final[str] = (
f'{os.path.join(self.MEDIA_DIR, tweet_id)}_{i}.ts'
)
mp4_filename: Final[str] = (
f'{os.path.join(self.MEDIA_DIR, tweet_id)}_{i}.mp4'
)
match self._download_m3u8(
media_path,
ts_filename,
mp4_filename,
accessor.proxies):
case FfmpegStatus.MP4:
logger.info(f'{mp4_filename}をアップロードしなければない。')
media_list.append(f'[[ファイル:{tweet_id}_{i}.mp4|240px]]')
case FfmpegStatus.TS:
logger.info(f'{ts_filename}.tsをmp4に変換してアップロードしなければない。')
media_list.append(f'[[ファイル:{tweet_id}_{i}.mp4|240px]]')
case FfmpegStatus.FAILED:
logger.error(f'{tweet_url}の動画が取得できませんでしたを 当職無能')
media_list.append('[[ファイル:(動画の取得ができませんでした)|240px]]')
case _ as unreachable: # pyright: ignore [reportUnnecessaryComparison]
assert_never(unreachable)
media_txt = ' '.join(media_list)
return media_txt
def _get_tweet_quote(self, tweet: Tag, accessor: AccessorHandler) -> str:
"""引用リツイートの引用元へのリンクを取得する。
Args:
tweet (Tag): ツイート内容である ``.timeline-item`` タグを表すbeautifulsoup4タグ。
accessor (AccessorHandler): アクセスハンドラ。
Returns:
str: Archiveテンプレートでラップされた引用元ツイートへのリンク。
"""
tweet_quote: Final[Tag | None] = tweet.select_one(
'.tweet-body > .quote.quote-big') # 引用リツイートを選択
quote_txt: str = ''
if tweet_quote is not None:
quote_link: Final[Tag | None] = (
tweet_quote.select_one('.quote-link'))
assert quote_link is not None
link_href: Final[str | list[str] | None] = quote_link.get('href')
assert isinstance(link_href, str)
link: str = self._url_fragment_pattern.sub('', link_href)
link = urljoin(self.TWITTER_URL, link)
quote_txt = self._archive_url(link, accessor)
tweet_quote_unavailable: Final[Tag | None] = tweet.select_one(
'.tweet-body > .quote.unavailable') # 引用リツイートを選択
if tweet_quote_unavailable is not None:
quote_txt = '(引用元が削除されました)'
return quote_txt
def _get_tweet_poll(self, tweet: Tag) -> str:
"""ツイートの投票結果を取得する。
Args:
tweet (Tag): ツイート内容である ``.timeline-item`` タグを表すbeautifulsoup4タグ。
Returns:
str: Wiki形式に書き直した投票結果。
"""
tweet_poll: Final[Tag | None] = tweet.select_one('.tweet-body > .poll')
poll_txt: str = ''
if tweet_poll is not None:
poll_meters: Final[ResultSet[Tag]] = tweet_poll.select(
'.poll-meter')
poll_info: Final[Tag | None] = tweet_poll.select_one('.poll-info')
assert poll_info is not None
for poll_meter in poll_meters:
poll_choice_value: Final[Tag | None] = poll_meter.select_one(
'.poll-choice-value')
assert poll_choice_value is not None
ratio: Final[str] = poll_choice_value.text
poll_choice_option: Final[Tag | None] = poll_meter.select_one(
'.poll-choice-option')
assert poll_choice_option is not None
if 'leader' in poll_meter['class']:
poll_txt += ('<br>\n'
' <span style="display: inline-block; '
'width: 30em; background: linear-gradient('
'to right, '
f'rgba(29, 155, 240, 0.58) 0 {ratio}, '
f'transparent {ratio} 100%); '
'font-weight: bold;">') \
+ ratio + ' ' + poll_choice_option.text + '</span>'
else:
poll_txt += ('<br>\n'
' <span style="display: inline-block; '
'width: 30em; background: linear-gradient('
'to right, '
f'rgb(207, 217, 222) 0 {ratio}, '
f'transparent {ratio} 100%);">') \
+ ratio + ' ' + poll_choice_option.text + '</span>'
poll_txt += '<br>\n <span style="font-size: small;">' \
+ poll_info.text + '</span>'
return poll_txt
def _get_timeline_items(self, soup: BeautifulSoup) -> list[Tag]:
"""タイムラインのツイートを取得。
基本的に投稿時刻の降順に取得し、リプライツリーは最後のツイートの時刻を基準として降順にひとまとまりにする。
Args:
soup (BeautifulSoup): Nitterのページを表すBeautifulSoupオブジェクト。
Returns:
list[Tag]: ツイートのアイテムである ``.timeline-item`` タグを表すTagオブジェクトのリスト。
"""
timeline_item_list: Final[list[Tag]] = []
for item_or_list in soup.select(
'.timeline > .timeline-item, .timeline > .thread-line'):
if 'unavailable' in item_or_list.attrs['class']:
continue
elif 'thread-line' in item_or_list.attrs['class']:
# そのままtimeline-itemクラスをfind_allするとツイートの順番が逆転するので、順番通りに取得するよう処理
for item in reversed(item_or_list.select('.timeline-item')):
timeline_item_list.append(item)
else:
timeline_item_list.append(item_or_list)
return timeline_item_list
def _archive_soup(self, tag: Tag, accessor: AccessorHandler) -> None:
"""ツイート内のaタグをテンプレートArchiveの文字列に変化させる。
NitterリンクをYouTubeへのリンクに、bibliogramへのリンクをInstagramへのリンクに修正する。
Args:
tag (Tag): ツイート内容の本文である ``.media-body`` タグを表すbeautifulsoup4タグ。
accessor (AccessorHandler): アクセスハンドラ。
"""
urls_in_tweet: Final[ResultSet[Tag]] = tag.find_all('a')
for url in urls_in_tweet:
href: Final[str | list[str] | None] = url.get('href')
assert isinstance(href, str)
if href.startswith('https://') or href.startswith('http://'):
# 先頭にhttpが付いていない物はハッシュタグの検索ページへのリンクなので処理しない
if href.startswith('https' + self.NITTER_INSTANCE[4:]):
# Nitter上のTwitterへのリンクを直す
url_link: str = href.replace(
'https' + self.NITTER_INSTANCE[4:], self.TWITTER_URL)
url_link = self._url_query_pattern.sub('', url_link)
url.replace_with(self._archive_url(url_link, accessor))
elif href.startswith('https://nitter.kavin.rocks/'):
# Nitter上のTwitterへのリンクを直す
url_link: str = href.replace(
'https://nitter.kavin.rocks/', self.TWITTER_URL)
url_link = self._url_query_pattern.sub('', url_link)
url.replace_with(self._archive_url(url_link, accessor))
elif (hasattr(self, '_invidious_pattern')
and self._invidious_pattern.search(href)):
# Nitter上のYouTubeへのリンクをInvidiousのものから直す
invidious_href: Final[str | list[str] | None] = (
self._invidious_pattern.sub(
'youtube.com' if (
re.match(r'https://[^/]+/[^/]+/', href)
or re.search(r'/@[^/]*$', href)
) else 'youtu.be',
href))
url.replace_with(self._archive_url(
invidious_href, accessor))
elif href.startswith('https://bibliogram.art/'):
# Nitter上のInstagramへのリンクをBibliogramのものから直す
# Bibliogramは中止されたようなのでそのうちリンクが変わるかも
url_link: str = href.replace(
'https://bibliogram.art/',
'https://www.instagram.com/')
url.replace_with(self._archive_url(url_link, accessor))
else:
url.replace_with(self._archive_url(href, accessor))
elif url.text.startswith('@'):
url_link: str = urljoin(self.TWITTER_URL, href)
url_text: Final[str] = url.text
url.replace_with(
self._archive_url(
url_link, accessor, url_text))
def _archive_url(
self,
url: str,
accessor: AccessorHandler,
text: str | None = None) -> str:
"""URLをArchiveテンプレートでラップする。
フラグメント識別子がURLに含まれていたら、Archive側のURLにも反映させる。
Args:
url (str): ラップするURL。
accessor (AccessorHandler): アクセスハンドラ。
text (str | None, optional): ArchiveテンプレートでURLの代わりに表示する文字列。
Returns:
str: ArchiveタグでラップしたURL。
"""
if '#' in url: # フラグメント識別子の処理
main_url, fragment = url.split('#', maxsplit=1)
return TableBuilder.archive_url(
url, self._archive(main_url, accessor) + '#' + fragment, text)
else:
return TableBuilder.archive_url(
url, self._archive(url, accessor), text)
def _callinshowlink_url(self, url: str, accessor: AccessorHandler) -> str:
"""URLをCallinShowLinkテンプレートでラップする。
Args:
url (str): ラップするURL。
accessor (AccessorHandler): アクセスハンドラ。
Returns:
str: CallinShowLinkタグでラップしたURL。
"""
return TableBuilder.callinshowlink_url(
url, self._archive(url, accessor))
def _archive(self, url: str, accessor: AccessorHandler) -> str:
"""URLから対応するarchive.todayのURLを返す。
取得できれば魚拓ページのURLを返す。
魚拓がなければその旨の警告を表示し、https://archive.ph/https%3A%2F%2Fxxxxxxxxの形で返される。
アクセスに失敗すればその旨の警告を表示し、https://archive.ph/https%3A%2F%2Fxxxxxxxxの形で返される。
Args:
url (str): 魚拓を取得するURL。
accessor (AccessorHandler): アクセスハンドラ。
Returns:
str: 魚拓のURL。
"""
archive_url: str = urljoin(
self.ARCHIVE_TODAY_STANDARD,
quote(unquote(url), safe='&=+?%'))
res: Final[str | None] = accessor.request(urljoin(
self.ARCHIVE_TODAY, quote(unquote(url), safe='&=+?%')))
if res is None: # 魚拓接続失敗時処理
# https://archive.ph/https%3A%2F%2Fxxxxxxxxの形で返される
logger.error(archive_url + 'にアクセス失敗ナリ。出力されるテキストにはそのまま記載されるナリ。')
else:
soup: Final[BeautifulSoup] = BeautifulSoup(res, 'html.parser')
content: Final[Tag | NavigableString | None] = soup.find(
id='CONTENT') # archive.todayの魚拓一覧ページの中身だけ取得
if (content is None or content.get_text()[:len(self.NO_ARCHIVE)]
== self.NO_ARCHIVE): # 魚拓があるかないか判定
logger.warning(url + 'の魚拓がない。これはいけない。')
else:
assert isinstance(content, Tag)
content_a: Final[Tag | NavigableString | None] = content.find(
'a')
assert isinstance(content_a, Tag)
href: Final[str | list[str] | None] = content_a.get('href')
assert isinstance(href, str)
archive_url = href.replace(
self.ARCHIVE_TODAY, self.ARCHIVE_TODAY_STANDARD)
return archive_url
def _get_tweet(self, accessor: AccessorHandler) -> bool:
"""ページからツイート本文を ``TableBuilder`` インスタンスに収めていく。
ツイートの記録を中断するための文を発見するか、記録したツイート数が :const:`~LIMIT_N_TWEETS` 件に達したら
`False` を返す。
Args:
accessor (AccessorHandler): アクセスハンドラ。
Returns:
bool: 終わりにするツイートを発見するか、記録件数が上限に達したら `False`。
"""
soup: Final[BeautifulSoup] = BeautifulSoup(
self._nitter_page, 'html.parser')
tweets: Final[list[Tag]] = self._get_timeline_items(soup)
for tweet in tweets:
tweet_a: Final[Tag | None] = tweet.a
assert tweet_a is not None
if tweet_a.text == self.NEWEST:
# Load Newestのボタンは処理しない
continue
if tweet.find(class_='retweet-header') is not None:
# retweet-headerはリツイートを示すので入っていれば処理しない
continue
if tweet.find(class_='pinned') is not None:
# pinnedは固定ツイートを示すので入っていれば処理しない
continue
if len(self._query_strs) > 0:
# クエリが指定されている場合、一つでも含まないツイートは処理しない
not_match: bool = False
for query_str in self._query_strs:
if query_str not in tweet.text:
not_match = True
break
if not_match:
continue
# 日付の更新処理
date: Final[datetime] = self._tweet_date(tweet)
self._table_builder.next_day_if_necessary(date)
tweet_link: Final[Tag | NavigableString | None] = tweet.find(
class_='tweet-link')
assert isinstance(tweet_link, Tag)
href: Final[str | list[str] | None] = tweet_link.get('href')
assert isinstance(href, str)
tweet_url: Final[str] = urljoin(
self.TWITTER_URL,
self._url_fragment_pattern.sub('', href))
tweet_callinshow_template: Final[str] = self._callinshowlink_url(
tweet_url, accessor)
tweet_content: Final[Tag | NavigableString | None] = tweet.find(
class_='tweet-content media-body')
assert isinstance(tweet_content, Tag)
self._archive_soup(tweet_content, accessor)
media_txt: Final[str] = self._fetch_tweet_media(
tweet, tweet_url, accessor)
quote_txt: Final[str] = self._get_tweet_quote(tweet, accessor)
poll_txt: Final[str] = self._get_tweet_poll(tweet)
self._table_builder.append(
tweet_callinshow_template, '<br>\n'.join(
filter(
None,
[
TableBuilder.escape_wiki_reserved_words(
tweet_content.get_text()),
quote_txt, media_txt, poll_txt
])))
if self._table_builder.count % self.REPORT_INTERVAL == 0:
logger.info(
f'ツイートを{self._table_builder.count}件も記録したンゴｗｗｗｗｗｗｗｗｗｗｗ')
if self._stop != '' and self._stop in tweet_content.get_text():
logger.info('目的ツイート発見でもう尾張屋根')
self._table_builder.dump_file()
return False
if self._table_builder.count >= self.LIMIT_N_TWEETS:
logger.info(f'{self.LIMIT_N_TWEETS}件も記録している。もうやめにしませんか。')
self._table_builder.dump_file()
return False
return True
def _go_to_new_page(self, accessor: AccessorHandler) -> bool:
"""Nitterで次のページに移動する。
次のページが無ければ `False` を返す。
Args:
accessor (AccessorHandler): アクセスハンドラ。
Returns:
bool: 次のページを取得できれば `True`。
"""
soup: Final[BeautifulSoup] = BeautifulSoup(self._nitter_page,
'html.parser')
show_mores: Final[ResultSet[Tag]] = soup.find_all(class_='show-more')
assert len(show_mores) > 0
new_url: str = ''
for show_more in show_mores: # show-moreに次ページへのリンクか前ページへのリンクがある
show_more_a: Final[Tag | None] = show_more.a
assert show_more_a is not None
href: Final[str | list[str] | None] = show_more_a.get('href')
assert isinstance(href, str)
new_url = urljoin(
self.NITTER_INSTANCE,
self._name
+ '/'
+ self.TWEETS_OR_REPLIES
+ href) # 直下のaタグのhrefの中身取ってURL頭部分と合体
res: Final[str | None] = accessor.request(new_url)
if res is None:
self._on_fail(accessor)
return False
new_page_soup: Final[BeautifulSoup] = BeautifulSoup(res, 'html.parser')
if new_page_soup.find(class_='timeline-end') is None:
# ツイートの終端ではtimeline-endだけのページになるので判定
logger.info(new_url + 'に移動しますを')
self._nitter_page = res # まだ残りツイートがあるのでページを返して再度ツイート本文収集
return True
else:
logger.info('急に残りツイートが無くなったな終了するか')
self._table_builder.dump_file()
return False
def execute(self, krsw: bool = False, use_browser: bool = True,
enable_javascript: bool = True) -> None | NoReturn:
"""通信が必要な部分のロジック。
Args:
krsw (bool, optional): `True` の場合、名前が自動で :const:`~CALLINSHOW` になり、
クエリと終わりにするツイートが自動で無しになる。
use_browser (bool, optional): `True` ならSeleniumを利用する。
`False` ならRequestsのみでアクセスする。
enable_javascript (bool, optional): SeleniumでJavaScriptを利用する場合は
`True`。
"""
# Seleniumドライバーを必ず終了するため、with文を利用する。
with AccessorHandler(use_browser, enable_javascript) as accessor:
# 実行前のチェック
self._check_nitter_instance(accessor)
self._check_archive_instance(accessor)
# Invidiousのインスタンスリストの正規表現パターンを取得
invidious_url_tuple: Final[tuple[str, ...]] = (
self._invidious_instances(accessor)
)
self._invidious_pattern: Pattern[str] = re.compile(
'|'.join(invidious_url_tuple))
# 検索クエリの設定
if not self._set_queries(accessor, krsw):
sys.exit(1)
# ツイートを取得し終えるまでループ
while True:
if not self._get_tweet(accessor):
break
if not self._go_to_new_page(accessor):
break
class UrlTuple(NamedTuple):
"""URLとその魚拓のURLのペア。
"""
url: str
"""URL。
"""
archive_url: str
"""魚拓のURL。
"""
class ArchiveCrawler(TwitterArchiver):
"""archive.todayに記録された尊師のツイートのうち、Wiki未掲載のものを収集する。
"""
TWEET_URL_PREFIX_DEFAULT: Final[str] = \
UserProperties.ArchiveCrawler.tweet_url_prefix_default
"""Final[str]: ツイートURLの数字部分のうち、予め固定しておく部分。
:func:`~_next_url` の `tweet_url_prefix` のデフォルト値。
"""
INCREMENTED_NUM_DEFAULT: Final[int] = \
UserProperties.ArchiveCrawler.incremented_num_default
"""Final[int]: ツイートURLの数字部分うち、インクリメントする桁のデフォルト値。
:const:`~TWEET_URL_PREFIX_DEFAULT` に続く桁をこの数字からインクリメントする。
:func:`~_next_url` の `incremented_num` のデフォルト値。
"""
TEMPLATE_URL: Final[str] = 'https://krsw-wiki.org/wiki/テンプレート:降臨ショー恒心ログ'
"""Final[str]: テンプレート:降臨ショー恒心ログのURL。
"""
FILENAME: Final[str] = UserProperties.ArchiveCrawler.filename
"""Final[str]: URLのリストをダンプするファイル名。
"""
@override
def _set_queries(self, accessor: AccessorHandler, krsw: bool) -> bool:
"""検索条件を設定する。
:class:`~TwitterArchiver` のメンバをセットし、ツイートを収集するアカウントを入力させる。
Args:
accessor (AccessorHandler): アクセスハンドラ
krsw (bool): `True` の場合、名前が :const:`~CALLINSHOW` になる。
"""
# ユーザー名取得
if krsw:
logger.info('名前は自動的に' + self.CALLINSHOW + 'にナリます')
self._name: str = self.CALLINSHOW
else:
name_optional: str | None = self._get_name(accessor)
if name_optional is not None:
self._name: str = name_optional
else:
return False
logger.info(
'ユーザー名: @' + self._name + 'で検索しまふ'
)
self._twitter_url_pattern: Pattern[str] = re.compile(
self.TWITTER_URL + self._name + r'/status/\d+')
self._url_list_on_wiki: list[str] = []
self._url_list: list[UrlTuple] = []
return True
def _get_tweet_urls_from_wiki(self, accessor: AccessorHandler) -> None:
"""Wikiに未掲載のツイートのURLリストを取得する。
Args:
accessor (AccessorHandler): アクセスハンドラ。
"""
def assert_get(tag: Tag, key: str) -> str:
"""BeautifulSoupのタグから属性値を取得する。
Args:
tag (Tag): BeautifulSoupのタグ。
key (str): 属性キー。
Returns:
str: タグの属性値。
"""
result: Final[str | list[str] | None] = tag.get(key)
assert isinstance(result, str)
return result
template_page: Final[str | None] = (
accessor.request_with_requests_module(self.TEMPLATE_URL))
assert template_page is not None
template_soup: Final[BeautifulSoup] = BeautifulSoup(
template_page,
'html.parser')
urls: Final[list[str]] = list(map(
lambda x: 'https://krsw-wiki.org' + assert_get(x, 'href'),
template_soup.select('.wikitable > tbody > tr > td a')))
for url in urls:
logger.info(f'{unquote(url)} で収集中でふ')
page: Final[str | None] = (
accessor.request_with_requests_module(url))
assert page is not None
soup: Final[BeautifulSoup] = BeautifulSoup(page, 'html.parser')
url_as = soup.select('tr > th a')
for url_a in url_as:
href: str | list[str] | None = url_a.get('href')
assert isinstance(href, str)
if href.startswith(self.TWITTER_URL + self._name):
self._url_list_on_wiki.append(href)
def _append_tweet_urls(self, soup: BeautifulSoup) -> None:
"""ツイートのURLを保存する。
Args:
soup (BeautifulSoup): archive.todayでのURL検索結果のページのオブジェクト。
"""
tweets: Final[ResultSet[Tag]] = soup.select(
'#CONTENT > div > .TEXT-BLOCK')
for tweet in tweets:
a_last_child: Final[Tag | None] = tweet.select_one('a:last-child')
assert a_last_child is not None
url_matched: Final[Match[str]] | None = (
self._twitter_url_pattern.match(a_last_child.text)
)
if url_matched is not None:
a_first_child: Final[Tag | None] = tweet.select_one(
'a:first-child')
assert a_first_child is not None
archive_url: Final[str | list[str] | None] = a_first_child.get(
'href')
assert isinstance(archive_url, str)
if url_matched[0] not in self._url_list_on_wiki:
# ツイートのURLが未取得のものならばURLを保存する
self._url_list.append(
UrlTuple(url_matched[0], archive_url))
self._url_list.sort(reverse=True, key=lambda x: x.url) # 降順
def _fetch_next_page(
self,
soup: BeautifulSoup,
accessor: AccessorHandler) -> str | None:
"""archive.todayの検索結果のページをpaginateする。
Args:
soup (BeautifulSoup): archive.todayでのURL検索結果のページのオブジェクト。
accessor (AccessorHandler): アクセスハンドラ。
Returns:
str | None: 次のページがあればそのHTML。
"""
next_a: Final[Tag | None] = soup.select_one('#next')
if next_a is not None:
link: Final[str | list[str] | None] = next_a.get('href')
assert isinstance(link, str)
page: Final[str | None] = accessor.request(link)
assert page is not None
return page
else:
return
def _get_tweet_loop(
self,
soup: BeautifulSoup,
accessor: AccessorHandler) -> None:
"""archive.todayの検索結果に対して、paginateしながら未記載のツイートURLを記録する。
Args:
soup (BeautifulSoup): archive.todayでのURL検索結果のページのオブジェクト。
accessor (AccessorHandler): アクセスハンドラ。
"""
has_next: bool = True
while has_next:
self._append_tweet_urls(soup)
next_page: Final[str | None] = self._fetch_next_page(
soup, accessor)
if next_page is not None:
soup = BeautifulSoup(next_page, 'html.parser')
else:
has_next = False
def _next_url(
self,
accessor: AccessorHandler,
tweet_url_prefix: str,
incremented_num: int) -> None:
"""ツイートのURLを、数字部分をインクリメントしながら探索する。
`https://twitter.com/CallinShow/status/` に続く数字部分について、
`tweet_url_prefix` で始まるものを、その次の桁を `incremented_num` から9までインクリメントして探索する。
Args:
accessor (AccessorHandler): アクセスハンドラ。
tweet_url_prefix (str): ツイートURLの数字部分のうち、インクリメントする桁以前の部分。
incremented_num (int): ツイートURLのうちインクリメントする桁の現在の数字。
Examples:
`https://twitter.com/CallinShow/status/1707` で始まるURLをすべて探索する場合
::
self._next_url(accessor, '1707', 0)
`https://twitter.com/CallinShow/status/165` で始まるURLから
`https://twitter.com/CallinShow/status/169` で始まるURLまでをすべて探索する場合
::
self._next_url(accessor, '16', 5)
See Also:
* :const:`~TWEET_URL_PREFIX_DEFAULT`: `tweet_url_prefix` のデフォルト値。
* :const:`~INCREMENTED_NUM_DEFAULT`: `incremented_num` のデフォルト値。
"""
assert 0 <= incremented_num and incremented_num <= 9, \
f'incremented_numが{incremented_num}でふ'
logger.info(self.TWITTER_URL + self._name + '/status/'
+ tweet_url_prefix + str(incremented_num) + '*を探索中')
page: Final[str | None] = accessor.request(
self.ARCHIVE_TODAY
+ self.TWITTER_URL
+ self._name
+ '/status/'
+ tweet_url_prefix
+ str(incremented_num) + '*')
assert page is not None
soup: Final[BeautifulSoup] = BeautifulSoup(page, 'html.parser')
pager: Final[Tag | None] = soup.select_one('#pager')
if pager is not None: # 検索結果が複数ページ
page_num_matched: Final[Match[str] | None] = re.search(
r'of (\d+) urls', pager.text)
assert page_num_matched is not None
page_num: Final[int] = int(page_num_matched[1])
if page_num > 100: # ツイート数が100を超えると途中でreCAPTCHAが入るので、もっと細かく検索
self._next_url(accessor,
tweet_url_prefix + str(incremented_num), 0)
else:
logger.debug(
self.TWITTER_URL + self._name + '/status/'
+ tweet_url_prefix + str(incremented_num) + '*からURLを収集しまふ')
self._get_tweet_loop(soup, accessor)
else: # 検索結果が1ページだけ
if soup.select_one('.TEXT-BLOCK'): # 検索結果が存在する場合
logger.debug(
self.TWITTER_URL + self._name + '/status/'
+ tweet_url_prefix + str(incremented_num) + '*からURLを収集しまふ')
self._get_tweet_loop(soup, accessor)
# 次のurlを探索
if incremented_num == 9:
return
else:
self._next_url(accessor, tweet_url_prefix, incremented_num + 1)
@override
def _tweet_date(self, tweet: Tag) -> datetime:
datetime_tag: Final[Tag | None] = tweet.select_one('time[datetime]')
assert datetime_tag is not None
datetime_str: Final[str | list[str] | None] = (
datetime_tag.get('datetime'))
assert isinstance(datetime_str, str)
raw_time: Final[datetime] = datetime.strptime(
datetime_str, '%Y-%m-%dT%H:%M:%SZ')
return raw_time.replace(tzinfo=ZoneInfo('Asia/Tokyo'))
def _filter_out_retweets(
self,
url_pairs: list[UrlTuple],
accessor: AccessorHandler) -> list[UrlTuple]:
"""リツイートを除く。
Args:
url_pairs (list[UrlTuple]): リツイートを除く前のURLのリスト。
accessor (AccessorHandler): アクセスハンドラ。
Returns:
list[UrlTuple]: リツイートを除いたURLのリスト。
"""
filtered_urls: Final[list[UrlTuple]] = []
for url_pair in url_pairs:
page: Final[str | None] = accessor.request(url_pair.archive_url)
assert page is not None
soup: Final[BeautifulSoup] = BeautifulSoup(page, 'html.parser')
if soup.select_one('span[data-testid="socialContext"]') is None:
filtered_urls.append(url_pair)
return filtered_urls
@override
def execute(self, krsw: bool = False, use_browser: bool = True,
enable_javascript: bool = True) -> None | NoReturn:
logger.info('Wikiに未掲載のツイートのURLを収集しますを')
# Seleniumドライバーを必ず終了するため、with文を利用する。
with AccessorHandler(use_browser, enable_javascript) as accessor:
# 実行前のチェック
self._check_archive_instance(accessor)
# 検索クエリの設定
if not self._set_queries(accessor, krsw):
sys.exit(1)
# Wikiに既に掲載されているツイートのURLを取得
self._get_tweet_urls_from_wiki(accessor)
# 未掲載のツイートのURLを取得する
self._next_url(accessor,
self.TWEET_URL_PREFIX_DEFAULT,
self.INCREMENTED_NUM_DEFAULT)
# リツイートを除く
filtered_url_list: Final[list[UrlTuple]] = (
self._filter_out_retweets(self._url_list, accessor))
with codecs.open(self.FILENAME, 'w', 'utf-8') as f:
for url_pair in filtered_url_list:
f.write(url_pair.url + '\n')
logger.info('テキストファイル手に入ったやで〜')
if __name__ == '__main__':
if sys.version_info < (3, 12):
print('Pythonのバージョンを3.12以上に上げて下さい')
logger.critical('貴職のPythonのバージョン: ' + str(sys.version_info))
sys.exit(1)
parser: Final[ArgumentParser] = ArgumentParser()
parser.add_argument(
'--krsw',
action='store_true',
help='指定すると、パカデブのツイートを取得上限数まで取得する。')
parser.add_argument(
'-n',
'--no-browser',
action='store_true',
help='指定すると、Tor Browserを利用しない。')
parser.add_argument(
'-d',
'--disable-script',
action='store_true',
help='指定すると、Tor BrowserでJavaScriptを利用しない。')
parser.add_argument(
'-u',
'--search-unarchived',
action='store_true',
help=('指定すると、Wikiに未掲載のツイートのURLをarchive.todayから収集する。'
'リツイートのURLも収集してしまうので注意。'))
args: Final[Namespace] = parser.parse_args()
logger.debug('args: ' + str(args))
twitter_archiver: Final[TwitterArchiver] = (
ArchiveCrawler() if args.search_unarchived else TwitterArchiver()
)
twitter_archiver.execute(
args.krsw,
not args.no_browser,
not args.disable_script)
実行例
20件での実行例。
12月10日
| https://twitter.com/CallinShow/status/1601539154256744449(魚拓) |
|---|
|
なお、ひめかファンのワタクシ、ブロックされております。 |
| https://twitter.com/CallinShow/status/1601542511302160384(魚拓) |
|
今日もお疲れ。 |
| https://twitter.com/CallinShow/status/1601569138379718656(魚拓) |
|
https://youtu.be/QR6Gj0MKcew(魚拓) |
| https://twitter.com/CallinShow/status/1601572951463428096(魚拓) |
|
菊地翔 |
| https://twitter.com/CallinShow/status/1601588268487041024(魚拓) |
|
日曜阪神11R 阪神JF |
12月11日
12月12日
| https://twitter.com/CallinShow/status/1601955353792430081(魚拓) |
|---|
|
エクシア合同会社のツイートの中に、岡ちゃんのツイートをリツイートしてしまってごめん。 |
| https://twitter.com/CallinShow/status/1601958600259371011(魚拓) |
|
日曜日に誰もエクシア合同会社のことなんか、呟きたいなんて思わないのが普通だと思う。 |
| https://twitter.com/CallinShow/status/1601959896252981249(魚拓) |
|
ときにはふざけたツイートをしているように思うかもしれない。 |
| https://twitter.com/CallinShow/status/1601961096528613376(魚拓) |
|
エクシア合同会社と闘う大人たちがやっていることは、Twitterを使った世直し運動なんだ。 |