唐澤貴洋Wiki

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

利用者ページ 議論

利用者:夜泣き/スクリプト (ソースを閲覧)

2023年9月23日 (土) 23:58時点における版

12,053 バイト追加 、 2023年9月23日 (土)
→‎コード: v4.0.0 最低限動くだけのものをとりとり
>Fet-Fe
(→‎コード: v3.0.3 機能に変化なし。機能の説明コメントをGoogleスタイルのdocstringに変更)
>Fet-Fe
(→‎コード: v4.0.0 最低限動くだけのものをとりとり)
7行目: 7行目:
"""Twitter自動収集スクリプト
"""Twitter自動収集スクリプト


ver3.0.3 2023/7/16恒心
ver4.0.0 2023/9/23恒心


当コードは恒心停止してしまった https://rentry.co/7298g の降臨ショーツイート自動収集スクリプトの復刻改善版です
当コードは恒心停止してしまった https://rentry.co/7298g の降臨ショーツイート自動収集スクリプトの復刻改善版です
33行目: 33行目:
     * MacOSの場合はbrewでtorコマンドを導入し、実行
     * MacOSの場合はbrewでtorコマンドを導入し、実行


   * PySocks, bs4はインストールしないと標準で入ってません
   * PySocks, bs4, seleniumはインストールしないと標準で入ってません
   * requestsも環境によっては入っていないかもしれない
   * requestsも環境によっては入っていないかもしれない


     * $ pip install bs4 requests PySocks
     * $ pip install bs4 requests PySocks selenium


   * pipも入っていなければ ``$ sudo apt install pip``
   * pipも入っていなければ ``$ sudo apt install pip``
49行目: 49行目:
import re
import re
import json
import json
import random
from datetime import datetime
from datetime import datetime
from zoneinfo import ZoneInfo
from zoneinfo import ZoneInfo
from time import sleep
from time import sleep
from typing import Final, NoReturn, TypeAlias
from types import MappingProxyType
from typing import Final, NoReturn, Any, Self
from collections.abc import Callable
from urllib.parse import quote, unquote, urljoin
from urllib.parse import quote, unquote, urljoin
import warnings
import warnings
import subprocess
import subprocess
import platform


import requests
import requests
import bs4
import bs4
from bs4 import BeautifulSoup
from bs4 import BeautifulSoup
from selenium import webdriver
from selenium.common.exceptions import NoSuchElementException, WebDriverException
from selenium.webdriver.firefox.options import Options
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.by import By


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


Response: TypeAlias = requests.models.Response
 
"""TypeAlias: requests.models.Responseの型エイリアス。
class AccessError(Exception):
"""
  """RequestとSeleniumで共通のアクセスエラー。
  """
  pass
 
 
class AbstractAccessor:
  """HTTPリクエストでWebサイトに接続するための基底クラス。
  """
 
  WAIT_TIME: Final[int] = 1
  """Final[int]: HTTPリクエスト成功失敗関わらず待機時間。
 
  1秒待つだけで行儀がいいクローラーだそうなので既定では1秒。
  しかし日本のポリホーモは1秒待っていても捕まえてくるので注意。
  https://ja.wikipedia.org/wiki/?curid=2187212
  """
 
  WAIT_RANGE: Final[int] = 5
  """Final[int]: ランダムな時間待機するときの待機時間の幅。
  """
 
  REQUEST_TIMEOUT: Final[int] = 30
  """Final[int]: HTTPリクエストのタイムアウト秒数。
  """
 
  def _random_sleep(self) -> None:
    """ランダムな秒数スリープする。
 
    自動操縦だとWebサイトに見破られないため。
    """
    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; rv:91.0) Gecko/20100101 Firefox/91.0'
  }
  """Final[dict[str, str]]: 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プロキシの設定。
  """
 
  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プロキシの設定。
  """
 
  TOR_CHECK_URL: Final[str] = 'https://check.torproject.org/api/ip'
  """Final[str]: Ｔor経由で通信しているかチェックするサイトのURL。
  """
 
  def __init__(self):
    """コンストラクタ。
    """
 
    self._proxies: dict[str, str] | None = None
    self._proxies = self._get_tor_proxies() ##Ｔorに必要なプロキシをセット
 
  def _execute(self, url: Final[str], proxies: dict[str, str]) -> requests.models.Response:
    """引数のURLにrequestsモジュールでHTTP接続する。
 
    Args:
      url Final[str]: 接続するURL。
      proxies dir[str, str]: 接続に利用するプロキシ。デフォルトでは :func:`~_get_tor_proxies` で設定した値を利用する。
 
    Returns:
      requests.models.Response: レスポンスのオブジェクト。
 
    Raises:
      requests.exceptions.ConnectionError: Ｔorで接続ができないなど、接続のエラー。
      AccessError: ステータスコードが200でない場合のエラー。
    """
    sleep(self.WAIT_TIME) ##DoS対策で待つ
    try:
      res: requests.models.Response = requests.get(url, timeout=self.REQUEST_TIMEOUT, headers=self.HEADERS, allow_redirects=False, proxies=proxies if proxies is not None else self._proxies)
      res.raise_for_status() ##HTTPステータスコードが200番台以外でエラー発生
    except requests.exceptions.ConnectionError:
      raise
    except requests.exceptions.RequestException as e:
      # requestsモジュール固有の例外を共通の例外に変換
      raise AccessError(str(e)) from None
    return res
 
  def get(self, url: Final[str], proxies: dict[str, str]=None) -> str:
    """引数のURLにrequestsモジュールでHTTP接続する。
 
    Args:
      url Final[str]: 接続するURL。
      proxies dir[str, str]: 接続に利用するプロキシ。デフォルトでは :func:`~_get_tor_proxies` で設定した値を利用する。
 
    Returns:
      str: レスポンスのHTML。
 
    Raises:
      requests.exceptions.ConnectionError: Ｔorで接続ができないなど、接続のエラー。
      AccessError: ステータスコードが200でない場合のエラー。
    """
    try:
      return self._execute(url, proxies).text
    except (requests.exceptions.ConnectionError, AccessError):
      raise
 
  def get_image(self, url: Final[str]) -> bytes | None:
    """引数のURLから画像のバイナリ列を取得する。
 
    Args:
      url Final[str]: 接続するURL
 
    Returns:
      bytes | None: 画像のバイナリ。画像でなければNone。
 
    Raises:
      requests.exceptions.ConnectionError: Ｔorで接続ができないなど、接続のエラー。
      AccessError: ステータスコードが200でない場合のエラー。
    """
    try:
      res: requests.models.Response = self._execute(url, self._proxies)
    except (requests.exceptions.ConnectionError, AccessError):
      raise
 
    if 'image' in res.headers['content-type']:
      return res.content
    else:
      return None
 
  def _get_tor_proxies(self) -> dict[str, str] | None | NoReturn:
    """Ｔorを使うのに必要なプロキシ情報を返す。
 
    プロキシなしで接続できればNone、Tor Browserのプロキシで接続できるなら :const:`~PROXIES_WITH_BROWSER`、torコマンドのプロキシで接続できるなら :const:`~PROXIES_WITH_COMMAND`。
    いずれでもアクセスできなければ異常終了する。
 
    Returns:
      dict[str, str] | None | NoReturn: プロキシ情報。
 
    Raises:
      RuntimeError: :const:`~TOR_CHECK_URL` にアクセスできない場合のエラー。
    """
    print('Ｔorのチェック中ですを')
    # プロキシなしでTorにアクセスできるかどうか
    res: str = self.get(self.TOR_CHECK_URL, proxies=None)
    is_tor: bool = json.loads(res)['IsTor']
    if is_tor:
      print('Ｔor connection OK')
      return None
 
    # Tor BrowserのプロキシでTorにアクセスできるかどうか
    res = self.get(self.TOR_CHECK_URL, proxies=self.PROXIES_WITH_BROWSER)
    is_tor = json.loads(res)['IsTor']
    if is_tor:
      print('Ｔor connection OK')
      return self.PROXIES_WITH_BROWSER
 
    # torコマンドのプロキシでTorにアクセスできるかどうか
    try:
      res = self.get(self.TOR_CHECK_URL, proxies=self.PROXIES_WITH_COMMAND)
      is_tor = json.loads(res)['IsTor']
      if is_tor:
        print('Ｔor proxy OK')
        return self.PROXIES_WITH_COMMAND
      else:
        raise RuntimeError('サイトにＴorのIPでアクセスできていないなりを')
    except requests.exceptions.ConnectionError as e:
      print(e, file=sys.stderr)
      print('通信がＴorのSOCKS proxyを経由していないなりを', file=sys.stderr)
      exit(1)
 
  @property
  def proxies(self) -> dict[str, str] | None:
    """オブジェクトのプロキシ設定を返す。
    """
    return self._proxies
 
 
class SeleniumAccessor(AbstractAccessor):
  """SeleniumでWebサイトに接続するためのクラス。
  """
 
  TOR_BROWSER_PATHS: MappingProxyType[str, str] = MappingProxyType({
      "Windows": "",
      "Darwin": "/Applications/Tor Browser.app/Contents/MacOS/firefox",
      "Linux": ""
  })
  """MappingProxyType[str, str]: OSごとのTor Browserのパス。
  """
 
  WAIT_TIME_FOR_INIT: Final[int] = 15
  """Final[int]: 最初のTor接続時の待機時間。
  """
 
  def __init__(self, enable_javascript: bool):
    """コンストラクタ。
 
    Tor Browserを自動操縦するためのSeleniumドライバを初期化する。
 
    Args:
      enable_javascript bool: JavaScriptを有効にするかどうか。reCAPTCHA対策に必要。
    """
 
    options: Options = Options()
    options.binary_location = self.TOR_BROWSER_PATHS[platform.system()]
 
    if enable_javascript:
      print("reCAPTCHA対策のためJavaScriptをonにしますを")
 
    options.preferences.update({
      "javascript.enabled": enable_javascript,
      "intl.accept_languages": "en-US, en",
      "intl.locale.requested": "US",
      "font.language.group": "x-western",
      "dom.webdriver.enabled": False # 自動操縦と見破られないための設定
    })
 
    self.driver: webdriver.Firefox = webdriver.Firefox(options=options)
    sleep(1)
    wait_init: WebDriverWait = WebDriverWait(self.driver, self.WAIT_TIME_FOR_INIT)
    wait_init.until(EC.element_to_be_clickable((By.ID, "connectButton")))
    self.driver.find_element(By.ID, "connectButton").click()
    wait_init.until(EC.url_contains("about:blank")) # Torの接続が完了するまで待つ
 
    self.wait: WebDriverWait = WebDriverWait(self.driver, self.REQUEST_TIMEOUT)
 
  def quit(self) -> None:
    """Seleniumドライバを終了する。
    """
    if self.driver:
      self.driver.quit()
 
  def _check_recaptcha(self) -> None:
    """reCAPTCHAが表示されているかどうか判定して、入力を待機する。
    """
    try:
      self.driver.find_element(By.CSS_SELECTOR, 'script[src^="https://www.google.com/recaptcha/api.js"]') # 要素がない時に例外を吐く
      print("reCAPTCHAを解いてね（笑）、それはできるよね。")
      print("botバレしたらNew Tor circuit for this siteを選択するナリよ")
      WebDriverWait(self.driver, 10000).until(EC.staleness_of(self.driver.find_element(By.CSS_SELECTOR, 'script[src^="https://www.google.com/recaptcha/api.js"]')))
      sleep(self.WAIT_TIME) ##DoS対策で待つ
    except NoSuchElementException:
      # reCAPTCHAの要素がなければそのまま
      pass
 
  def get(self, url: Final[str]) -> str:
    """引数のURLにSeleniumでHTTP接続する。
 
    Args:
      url Final[str]: 接続するURL。
 
    Returns:
      str: レスポンスのHTML。
    """
    self._random_sleep() ##DoS対策で待つ
    try:
      self.driver.get(url)
      self._check_recaptcha()
    except WebDriverException as e:
      # Selenium固有の例外を共通の例外に変換
      raise AccessError(str(e)) from None
    return self.driver.page_source
 
 
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=True, enable_javascript=True):
    """コンストラクタ。
 
    Requestのみを利用するか、Seleniumも利用するか引数で選択して初期化する。
 
    Args:
      use_browser bool: TrueならSeleniumを利用する。FalseならRequestsのみでアクセスする。
      enable_javascript bool: SeleniumでJavaScriptを利用する場合はTrue。
    """
    self.selenium_accessor: SeleniumAccessor | None = SeleniumAccessor(enable_javascript) if use_browser else None
    self.requests_accessor: RequestsAccessor = RequestsAccessor()
 
  def __enter__(self) -> Self:
    """withブロックの開始時に実行する。
    """
    return self
 
  def __exit__(self, *args) -> None:
    """withブロックの終了時に実行する。
    """
    if self.selenium_accessor is not None:
      self.selenium_accessor.quit()
 
  def request_once(self, url: Final[str]) -> str:
    """引数のURLにHTTP接続する。
 
    Args:
      url Final[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(self, url: Final[str], request_callable: Callable[[str], Any]) -> Any | None:
    """request_callableの実行を :const:`~LIMIT_N_REQUESTS` で定義された回数まで試す。
 
    成功すると結果を返す。
    接続失敗が何度も起きるとNoneを返す。
 
    Args:
      url Final[str]: 接続するURL
      request_callable Callable[[str], Any]: 1回リクエストを行うメソッド。
 
    Returns:
      Any | None: レスポンス。接続失敗が何度も起きるとNoneを返す。
 
    Note:
      失敗かどうかは呼出側で要判定
    """
    for i in range(self.LIMIT_N_REQUESTS):
      try:
        res: Any = request_callable(url)
      except AccessError as e:
        print(url + 'への通信失敗ナリ  ' + f"{i}/{self.LIMIT_N_REQUESTS}回")
        sleep(self.WAIT_TIME_FOR_ERROR) ##失敗時は長めに待つ
      else:
        return res
    return None
 
  def request(self, url: Final[str]) -> str | None:
    """HTTP接続を :const:`~LIMIT_N_REQUESTS` で定義された回数まで試す。
 
    成功すると結果を返す。
    接続失敗が何度も起きるとNoneを返す。
 
    Args:
      url Final[str]: 接続するURL
 
    Returns:
      str | None: レスポンスのテキスト。接続失敗が何度も起きるとNoneを返す。
 
    Note:
      失敗かどうかは呼出側で要判定
    """
    return self._request_with_callable(url, self.request_once)
 
  def request_with_requests_module(self, url: Final[str]) -> str | None:
    """requestsモジュールでのHTTP接続を :const:`~LIMIT_N_REQUESTS` で定義された回数まで試す。
 
    成功すると結果を返す。
    接続失敗が何度も起きるとNoneを返す。
 
    Args:
      url Final[str]: 接続するURL
 
    Returns:
      str | None: レスポンスのテキスト。接続失敗が何度も起きるとNoneを返す。
 
    Note:
      失敗かどうかは呼出側で要判定
    """
    return self._request_with_callable(url, self.requests_accessor.get)
 
  def request_image(self, url: Final[str]) -> bytes | None:
    """requestsモジュールで画像ファイルを取得します。
 
    成功すると結果を返します。
    接続失敗が何度も起きるとNoneを返します。
 
    Args:
      url Final[str]: 接続するURL
 
    Returns:
      bytes | None: レスポンスのバイト列。接続失敗が何度も起きるとNoneを返します。
    """
    return self._request_with_callable(url, self.requests_accessor.get_image)
 
  @property
  def proxies(self) -> dict[str, str] | None:
    return self.requests_accessor.proxies
 


class TwitterArchiver:
class TwitterArchiver:
75行目: 491行目:
   """
   """


  #定数・設定類
   NITTER_INSTANCE: Final[str] = 'https://nitter.net/'
 
   NITTER_INSTANCE: Final[str] = 'http://nitter7bryz3jv7e3uekphigvmoyoem4al3fynerxkj22dmoxoq553qd.onion/'
   """Final[str]: Nitterのインスタンス。
   """Final[str]: Nitterのインスタンス。


84行目: 498行目:
   Note:
   Note:
     末尾にスラッシュ必須。
     末尾にスラッシュ必須。
  Todo:
    Tor専用のインスタンスが使えるようになったら変更する。
   """
   """


90行目: 507行目:


   ページにアクセスして魚拓があるか調べるのにはonion版のarchive.todayを使用。
   ページにアクセスして魚拓があるか調べるのにはonion版のarchive.todayを使用。
   Note:
   Note:
     末尾にスラッシュ必須。
     末尾にスラッシュ必須。
113行目: 529行目:
   CALLINSHOW: Final[str] = 'CallinShow'
   CALLINSHOW: Final[str] = 'CallinShow'
   """Final[str]: 降臨ショーのユーザーネーム。
   """Final[str]: 降臨ショーのユーザーネーム。
  """
  REQUEST_TIMEOUT: Final[int] = 30
  """Final[int]: HTTPリクエストのタイムアウト秒数。
   """
   """


125行目: 537行目:
   REPORT_INTERVAL: Final[int] = 5
   REPORT_INTERVAL: Final[int] = 5
   """Final[int]: 記録件数を報告するインターバル。
   """Final[int]: 記録件数を報告するインターバル。
  """
  LIMIT_N_REQUESTS: Final[int] = 5
  """Final[int]: HTTPリクエスト失敗時の再試行回数。
  """
  WAIT_TIME_FOR_ERROR: Final[int] = 4
  """Final[int]: HTTPリクエスト失敗時にさらに追加する待機時間。
  """
  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のドメインとユーザーネーム部分の接続部品。
   """Final[str]: NitterのURLのドメインとユーザーネーム部分の接続部品。
  """
  HEADERS: Final[dict[str, str]] = {
    '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]] = {
    'http': 'socks5h://127.0.0.1:9050',
    'https': 'socks5h://127.0.0.1:9050'
  }
  """Final[dict[str, str]]: HTTPプロキシの設定。
   """
   """


184行目: 567行目:
   def __init__(self, krsw: bool=False):
   def __init__(self, krsw: bool=False):
     """コンストラクタ
     """コンストラクタ
    """
    self._txt_data: list[str] = []
    self._limit_count: int = 0 ##記録数
  def _set_queries(self, accessor: AccessorHandler, krsw: bool):
    """検索条件を設定する。


     :class:`~TwitterArchiver` のメンバをセットし、ツイートを収集するアカウントと
     :class:`~TwitterArchiver` のメンバをセットし、ツイートを収集するアカウントと
189行目: 578行目:


     Args:
     Args:
      accessor AccessorHandler: アクセスハンドラ
       krsw bool: Trueの場合、名前が自動で :const:`~CALLINSHOW` になり、クエリと終わりにするツイートが自動で無しになる。
       krsw bool: Trueの場合、名前が自動で :const:`~CALLINSHOW` になり、クエリと終わりにするツイートが自動で無しになる。
     """
     """
    self._txt_data: list[str] = []
    self._limit_count: int = 0 ##記録数
    self._proxy_is_needed: bool = False ## アクセスにプロキシが必要かどうか
    self._check_slash() ##スラッシュが抜けてないかチェック
    self._proxy_is_needed = self._check_tor_proxy_is_needed() ##Ｔorが使えているかチェック
    self._check_nitter_instance() ##インスタンスが死んでないかチェック
    self._check_archive_instance()
    ##Invidiousのインスタンスリストの正規表現パターンを取得
    invidious_url_tuple: Final[tuple[str]] = self._invidious_instances()
    self._invidious_pattern: Final[re.Pattern] = re.compile('|'.join(invidious_url_tuple))


     ##ユーザー名取得
     ##ユーザー名取得
208行目: 587行目:
       self.name: Final[str] = self.CALLINSHOW
       self.name: Final[str] = self.CALLINSHOW
     else:
     else:
       self.name: Final[str] = self._get_name()
       self.name: Final[str] = self._get_name(accessor)


     ##検索クエリとページ取得
     ##検索クエリとページ取得
216行目: 595行目:
     else:
     else:
       self._get_query()
       self._get_query()
     self._page: Response | None = self._request(urljoin(self.NITTER_INSTANCE, self.name + '/' + self.TWEETS_OR_REPLIES))
     self._page: Final[str] | None = accessor.request(urljoin(self.NITTER_INSTANCE, self.name + '/' + self.TWEETS_OR_REPLIES))
     if self._page is None:
     if self._page is None:
       self._fail()
       self._fail()
228行目: 607行目:


     ##日付取得
     ##日付取得
     self._date: datetime = self._tweet_date(BeautifulSoup(self._page.text, 'html.parser').find(class_='timeline-item'))
     self._date: datetime = self._tweet_date(BeautifulSoup(self._page, 'html.parser').find(class_='timeline-item'))
     self._txt_data.append('')
     self._txt_data.append('')
     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:
      res: Response = requests.get(url, timeout=self.REQUEST_TIMEOUT, headers=self.HEADERS, allow_redirects=False, proxies=self.PROXIES)
    else:
      res: Response = requests.get(url, timeout=self.REQUEST_TIMEOUT, headers=self.HEADERS, allow_redirects=False)
    sleep(self.WAIT_TIME) ##DoS対策で待つ
    return res
  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 ##リクエスト挑戦回数を記録
    while True:
      try:
        res: Response = self._request_once(url) ##リクエスト
        res.raise_for_status() ##HTTPステータスコードが200番台以外でエラー発生
      except requests.exceptions.RequestException as e:
        print(url + 'への通信失敗ナリ  ' + f"{counter}/{self.LIMIT_N_REQUESTS}回")
        if counter < self.LIMIT_N_REQUESTS: ##エラー発生時上限まで再挑戦
          counter += 1 ##現在の試行回数1回増やす
          sleep(self.WAIT_TIME_FOR_ERROR) ##失敗時は長めに待つ
        else:
          return None ##失敗したらNone返却し呼出側で対処してもらう
      else:
        return res ##リクエストの結果返す


   def _check_slash(self) -> None | NoReturn:
   def _check_slash(self) -> None | NoReturn:
299行目: 629行目:
       raise RuntimeError('TWITTER_URLの末尾には/が必須です')
       raise RuntimeError('TWITTER_URLの末尾には/が必須です')


   def _check_tor_proxy_is_needed(self) -> bool | NoReturn:
   def _check_nitter_instance(self, accessor: AccessorHandler) -> None | NoReturn:
    """Ｔorが使えているかチェックします。
 
    Returns:
      bool | NoReturn: Ｔor用のプロキシを通さなくてもＴor通信になっていればFalse。プロキシを通す必要があればTrue。
 
    Raises:
      RuntimeError: https://check.torproject.org/api/ip にアクセスできなければ出る。
    """
    initial_proxy_is_needed: bool = self._proxy_is_needed
    print('Ｔorのチェック中ですを')
    # プロキシなしでTorにアクセスできるかどうか
    self._proxy_is_needed = False
    res: Response = self._request_once('https://check.torproject.org/api/ip') ##リクエスト
    is_tor: bool = json.loads(res.text)['IsTor']
    if is_tor:
      print('Ｔor connection OK')
      self._proxy_is_needed = initial_proxy_is_needed
      return False
 
    # プロキシありでTorにアクセスできるかどうか
    self._proxy_is_needed = True
    try:
      res = self._request_once('https://check.torproject.org/api/ip') ##リクエスト
      is_tor = json.loads(res.text)['IsTor']
      if is_tor:
        print('Ｔor proxy OK')
        self._proxy_is_needed = initial_proxy_is_needed
        return True
      else:
        raise RuntimeError('サイトにＴorのIPでアクセスできていないなりを')
    except requests.exceptions.ConnectionError as e:
      print(e, file=sys.stderr)
      print('通信がＴorのSOCKS proxyを経由していないなりを', file=sys.stderr)
      exit(1)
 
  def _check_nitter_instance(self) -> None | NoReturn:
     """Nitterのインスタンスが生きているかチェックする。
     """Nitterのインスタンスが生きているかチェックする。


     死んでいたらそこで終了。
     死んでいたらそこで終了。
     接続を一回しか試さない :func:`~_request_once` を使っているのは、激重インスタンスが指定されたとき試行回数増やして偶然成功してそのまま実行されるのを躱すため。
     接続を一回しか試さない :func:`~_request_once` を使っているのは、激重インスタンスが指定されたとき試行回数増やして偶然成功してそのまま実行されるのを躱すため。
    Args:
      accessor AccessorHandler: アクセスハンドラ


     Returns:
     Returns:
346行目: 643行目:
     print("Nitterのインスタンスチェック中ですを")
     print("Nitterのインスタンスチェック中ですを")
     try:
     try:
       res: Final[Response] = self._request_once(self.NITTER_INSTANCE) ##リクエスト
       accessor.request_once(self.NITTER_INSTANCE)
      res.raise_for_status() ##HTTPステータスコードが200番台以外でエラー発生
     except AccessError as e: ##エラー発生時は終了
     except requests.exceptions.RequestException as e: ##エラー発生時は終了
       print(e, file=sys.stderr)
       print(e, file=sys.stderr)
       print('インスタンスが死んでますを', file=sys.stderr)
       print('インスタンスが死んでますを', file=sys.stderr)
354行目: 650行目:
     print("Nitter OK")
     print("Nitter OK")


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


     Returns:
     Returns:
362行目: 661行目:
     print("archive.todayのＴorインスタンスチェック中ですを")
     print("archive.todayのＴorインスタンスチェック中ですを")
     try:
     try:
       res: Final[Response] = self._request_once(self.ARCHIVE_TODAY) ##リクエスト
       accessor.request_once(self.ARCHIVE_TODAY)
      res.raise_for_status() ##HTTPステータスコードが200番台以外でエラー発生
     except AccessError as e: ##エラー発生時は終了
     except requests.exceptions.RequestException as e: ##エラー発生時は終了
       print(e, file=sys.stderr)
       print(e, file=sys.stderr)
       print('インスタンスが死んでますを', file=sys.stderr)
       print('インスタンスが死んでますを', file=sys.stderr)
370行目: 668行目:
     print("archive.today OK")
     print("archive.today OK")


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


     Returns:
     Returns:
377行目: 678行目:
     """
     """
     print("Invidiousのインスタンスリストを取得中ですを")
     print("Invidiousのインスタンスリストを取得中ですを")
     invidious_json: Response | None = self._request('https://api.invidious.io/instances.json')
     invidious_json: Final[str] | None = accessor.request_with_requests_module('https://api.invidious.io/instances.json')
     if invidious_json is None:
     if invidious_json is None:
       print("Invidiousが死んでますを")
       print("Invidiousが死んでますを")
       exit(1)
       exit(1)
     instance_list: list[str] = []
     instance_list: list[str] = []
     for instance_info in json.loads(invidious_json.text):
     for instance_info in json.loads(invidious_json):
       instance_list.append(instance_info[0])
       instance_list.append(instance_info[0])


392行目: 693行目:
     return tuple(instance_list)
     return tuple(instance_list)


 
   def _get_name(self, accessor: AccessorHandler) -> str | NoReturn:
   def _get_name(self) -> str | NoReturn:
     """ツイート収集するユーザー名を標準入力から取得する。
     """ツイート収集するユーザー名を標準入力から取得する。


     何も入力しないと :const:`~CALLINSHOW` を指定する。
     何も入力しないと :const:`~CALLINSHOW` を指定する。
    Args:
      accessor AccessorHandler: アクセスハンドラ


     Returns:
     Returns:
408行目: 711行目:
         return self.CALLINSHOW
         return self.CALLINSHOW
       else:
       else:
         res: Response | None = self._request(urljoin(self.NITTER_INSTANCE, account_str)) ##リクエストして結果取得
         res: Final[str]| None = accessor.request(urljoin(self.NITTER_INSTANCE, account_str)) ##リクエストして結果取得
         if res is None : ##リクエスト失敗判定
         if res is None : ##リクエスト失敗判定
           self._fail()
           self._fail()
         soup: BeautifulSoup = BeautifulSoup(res.text, 'html.parser') ##beautifulsoupでレスポンス解析
         soup: BeautifulSoup = BeautifulSoup(res, 'html.parser') ##beautifulsoupでレスポンス解析
         if soup.title == self.NITTER_ERROR_TITLE: ##タイトルがエラーでないか判定
         if soup.title == self.NITTER_ERROR_TITLE: ##タイトルがエラーでないか判定
           print(account_str + "は実在の人物ではありませんでした") ##エラー時ループに戻る
           print(account_str + "は実在の人物ではありませんでした") ##エラー時ループに戻る
476行目: 779行目:
     return end_str
     return end_str


   def _download_media(self, media_name: Final[str]) -> bool:
   def _download_media(self, media_name: Final[str], accessor: AccessorHandler) -> bool:
     """ツイートの画像をダウンロードし、:const:`~MEDIA_DIR` に保存する。
     """ツイートの画像をダウンロードし、:const:`~MEDIA_DIR` に保存する。


     Args:
     Args:
       media_name Final[str]: 画像ファイル名。Nitter上のimgタグのsrc属性では、``/pic/media%2F`` に後続する。
       media_name Final[str]: 画像ファイル名。Nitter上のimgタグのsrc属性では、``/pic/media%2F`` に後続する。
      accessor AccessorHandler: アクセスハンドラ


     Returns:
     Returns:
487行目: 791行目:
     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)
     res: Final[Response | None] = self._request(url)
     image_bytes: Final[bytes | None] = accessor.request_image(url)
     if res is not None:
     if image_bytes is not None:
      if 'image' not in res.headers['content-type']:
        return False
       with open(os.path.join(self.MEDIA_DIR, media_name), "wb") as f:
       with open(os.path.join(self.MEDIA_DIR, media_name), "wb") as f:
         f.write(res.content)
         f.write(image_bytes)
       return True
       return True
     else:
     else:
530行目: 832行目:
       self._date = date
       self._date = date


   def _get_tweet_media(self, tweet: bs4.element.Tag) -> str:
   def _get_tweet_media(self, tweet: bs4.element.Tag, accessor: AccessorHandler) -> str:
     """ツイートの画像や動画を取得する。
     """ツイートの画像や動画を取得する。


     Args:
     Args:
       tweet bs4.element.Tag: ツイート内容である ``.timeline-item`` タグを表すbeautifulsoup4タグ。
       tweet bs4.element.Tag: ツイート内容である ``.timeline-item`` タグを表すbeautifulsoup4タグ。
      accessor AccessorHandler: アクセスハンドラ


     Returns:
     Returns:
548行目: 851行目:
           media_name: str = [group for group in re.search(r'%2F([^%]*\.jpg)|%2F([^%]*\.jpeg)|%2F([^%]*\.png)|%2F([^%]*\.gif)', image_a.get('href')).groups() if group][0]
           media_name: str = [group for group in re.search(r'%2F([^%]*\.jpg)|%2F([^%]*\.jpeg)|%2F([^%]*\.png)|%2F([^%]*\.gif)', image_a.get('href')).groups() if group][0]
           media_list.append(f"[[ファイル:{media_name}|240px]]")
           media_list.append(f"[[ファイル:{media_name}|240px]]")
           if self._download_media(media_name):
           if self._download_media(media_name, accessor):
             print(os.path.join(self.MEDIA_DIR, media_name) + ' をアップロードしなければない。')
             print(os.path.join(self.MEDIA_DIR, media_name) + ' をアップロードしなければない。')
           else:
           else:
557行目: 860行目:
           media_list.append(f"[[ファイル:(画像の取得ができませんでした)|240px]]")
           media_list.append(f"[[ファイル:(画像の取得ができませんでした)|240px]]")
       # ツイートの動画の取得。ffmpegが入っていなければ手動で取得すること
       # ツイートの動画の取得。ffmpegが入っていなければ手動で取得すること
       for i, video in enumerate(tweet_media.select('.attachment.video-container video')):
       for i, video_container in enumerate(tweet_media.select('.attachment.video-container')):
         tweet_url: str = urljoin(self.TWITTER_URL, re.sub('#[^#]*$', '', tweet.find(class_='tweet-link').get('href'))) ##ツイートのURL作成
         tweet_url: str = urljoin(self.TWITTER_URL, re.sub('#[^#]*$', '', tweet.find(class_='tweet-link').get('href'))) ##ツイートのURL作成
        video = video_container.select_one('video')
        if video is None:
          print(f"{tweet_url}の動画が取得できませんでしたを 当職無能")
          media_list.append(f"[[ファイル:(動画の取得ができませんでした)|240px]]")
          continue
         if subprocess.run(['which', 'ffmpeg'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL).returncode != 0:
         if subprocess.run(['which', 'ffmpeg'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL).returncode != 0:
           print(f"ffmpegがないため{tweet_url}の動画が取得できませんでしたを")
           print(f"ffmpegがないため{tweet_url}の動画が取得できませんでしたを")
566行目: 875行目:
           tweet_id: str = tweet_url.split('/')[-1]
           tweet_id: str = tweet_url.split('/')[-1]
           # 動画のダウンロード
           # 動画のダウンロード
           if self._proxy_is_needed:
           if accessor.proxies is not None:
               returncode: int = subprocess.run(["ffmpeg", "-y", "-http_proxy", "self.PROXIES['http']", "-i", urljoin(self.NITTER_INSTANCE, media_url), "-c", "copy", f"{os.path.join(self.MEDIA_DIR, tweet_id)}_{i}.ts"], stdout=subprocess.DEVNULL).returncode
               returncode: int = subprocess.run(["ffmpeg", "-y", "-http_proxy", "accessor.proxies['http']", "-i", urljoin(self.NITTER_INSTANCE, media_url), "-c", "copy", f"{os.path.join(self.MEDIA_DIR, tweet_id)}_{i}.ts"], stdout=subprocess.DEVNULL).returncode
           else:
           else:
               returncode: int = subprocess.run(["ffmpeg", "-y", "-i", urljoin(self.NITTER_INSTANCE, media_url), "-c", "copy", f"{os.path.join(self.MEDIA_DIR, tweet_id)}_{i}.ts"], stdout=subprocess.DEVNULL).returncode
               returncode: int = subprocess.run(["ffmpeg", "-y", "-i", urljoin(self.NITTER_INSTANCE, media_url), "-c", "copy", f"{os.path.join(self.MEDIA_DIR, tweet_id)}_{i}.ts"], stdout=subprocess.DEVNULL).returncode
584行目: 893行目:
     return media_txt
     return media_txt


   def _get_tweet_quote(self, tweet: bs4.element.Tag) -> str:
   def _get_tweet_quote(self, tweet: bs4.element.Tag, accessor: AccessorHandler) -> str:
     """引用リツイートの引用元へのリンクを取得する。
     """引用リツイートの引用元へのリンクを取得する。


     Args:
     Args:
       tweet bs4.element.Tag: ツイート内容である ``.timeline-item`` タグを表すbeautifulsoup4タグ。
       tweet bs4.element.Tag: ツイート内容である ``.timeline-item`` タグを表すbeautifulsoup4タグ。
      accessor AccessorHandler: アクセスハンドラ


     Returns:
     Returns:
599行目: 909行目:
       link = re.sub('#.*$', '', link)
       link = re.sub('#.*$', '', link)
       link = urljoin(self.TWITTER_URL, link)
       link = urljoin(self.TWITTER_URL, link)
       quote_txt = self._archive_url(link)
       quote_txt = self._archive_url(link, accessor)
     tweet_quote_unavailable: Final[bs4.element.Tag | None] = tweet.select_one('.tweet-body > .quote.unavailable') # 引用リツイートを選択
     tweet_quote_unavailable: Final[bs4.element.Tag | None] = tweet.select_one('.tweet-body > .quote.unavailable') # 引用リツイートを選択
     if tweet_quote_unavailable is not None:
     if tweet_quote_unavailable is not None:
650行目: 960行目:
     return timeline_item_list
     return timeline_item_list


 
   def _get_tweet(self, accessor: AccessorHandler) -> None | NoReturn:
   def get_tweet(self) -> None | NoReturn:
     """ページからツイート本文を ``self._txt_data`` に収めていく。
     """ページからツイート本文を ``self._txt_data`` に収めていく。


     ツイートの記録を中断するための文を発見するか、記録したツイート数が :const:`~LIMIT_N_TWEETS` 件に達したら終了する。
     ツイートの記録を中断するための文を発見するか、記録したツイート数が :const:`~LIMIT_N_TWEETS` 件に達したら終了する。
    Args:
      accessor AccessorHandler: アクセスハンドラ
     """
     """
     soup: Final[BeautifulSoup] = BeautifulSoup(self._page.text, 'html.parser') ##beautifulsoupでレスポンス解析
     soup: Final[BeautifulSoup] = BeautifulSoup(self._page, '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) ##一ツイートのブロックごとにリストで取得
     for tweet in tweets: ##一ツイート毎に処理
     for tweet in tweets: ##一ツイート毎に処理
678行目: 990行目:
       if date.year != self._date.year or date.month != self._date.month or date.day != self._date.day:
       if date.year != self._date.year or date.month != self._date.month or date.day != self._date.day:
         self._next_day(date)
         self._next_day(date)
       archived_tweet_url: str = self._callinshowlink_url(tweet_url) ##ツイートURLをテンプレートCallinShowlinkに変化
       archived_tweet_url: str = self._callinshowlink_url(tweet_url, accessor) ##ツイートURLをテンプレートCallinShowlinkに変化
       tweet_content: bs4.element.Tag = tweet.find(class_='tweet-content media-body') ##ツイートの中身だけ取り出す
       tweet_content: bs4.element.Tag = tweet.find(class_='tweet-content media-body') ##ツイートの中身だけ取り出す
       self._archive_soup(tweet_content) ##ツイートの中身のリンクをテンプレートArchiveに変化
       self._archive_soup(tweet_content, accessor) ##ツイートの中身のリンクをテンプレートArchiveに変化
       media_txt: str = self._get_tweet_media(tweet) ##ツイートに画像などのメディアを追加
       media_txt: str = self._get_tweet_media(tweet, accessor) ##ツイートに画像などのメディアを追加
       quote_txt: str = self._get_tweet_quote(tweet) ##引用リツイートの場合、元ツイートを追加
       quote_txt: str = self._get_tweet_quote(tweet, accessor) ##引用リツイートの場合、元ツイートを追加
       poll_txt: str = self._get_tweet_poll(tweet) ##投票の取得
       poll_txt: str = self._get_tweet_poll(tweet) ##投票の取得
       self._txt_data[0] = '!' + archived_tweet_url + '\n|-\n|\n' \
       self._txt_data[0] = '!' + archived_tweet_url + '\n|-\n|\n' \
748行目: 1,060行目:
     return text
     return text


   def _archive_soup(self, tag: bs4.element.Tag) -> None:
   def _archive_soup(self, tag: bs4.element.Tag, accessor: AccessorHandler) -> None:
     """ツイート内のaタグをテンプレートArchiveの文字列に変化させる。
     """ツイート内のaタグをテンプレートArchiveの文字列に変化させる。


755行目: 1,067行目:
     Args:
     Args:
       tag bs4.element.Tag: ツイート内容の本文である ``.media-body`` タグを表すbeautifulsoup4タグ。
       tag bs4.element.Tag: ツイート内容の本文である ``.media-body`` タグを表すbeautifulsoup4タグ。
      accessor AccessorHandler: アクセスハンドラ
     """
     """
     urls_in_tweet: Final[bs4.element.ResultSet] = tag.find_all('a')
     urls_in_tweet: Final[bs4.element.ResultSet] = tag.find_all('a')
763行目: 1,076行目:
           url_link: str = url.get('href').replace('https' + self.NITTER_INSTANCE[4:], self.TWITTER_URL)
           url_link: str = url.get('href').replace('https' + self.NITTER_INSTANCE[4:], self.TWITTER_URL)
           url_link = re.sub('\?.*$', '', url_link)
           url_link = re.sub('\?.*$', '', url_link)
           url.replace_with(self._archive_url(url_link)) ##テンプレートArchiveに変化
           url.replace_with(self._archive_url(url_link, accessor)) ##テンプレートArchiveに変化
         elif url.get('href').startswith('https://nitter.kavin.rocks/'):
         elif url.get('href').startswith('https://nitter.kavin.rocks/'):
           #Nitter上のTwitterへのリンクを直す
           #Nitter上のTwitterへのリンクを直す
           url_link: str = url.get('href').replace('https://nitter.kavin.rocks/', self.TWITTER_URL)
           url_link: str = url.get('href').replace('https://nitter.kavin.rocks/', self.TWITTER_URL)
           url_link = re.sub('\?.*$', '', url_link)
           url_link = re.sub('\?.*$', '', url_link)
           url.replace_with(self._archive_url(url_link)) ##テンプレートArchiveに変化
           url.replace_with(self._archive_url(url_link, accessor)) ##テンプレートArchiveに変化
         elif self._invidious_pattern.search(url.get('href')):
         elif self._invidious_pattern.search(url.get('href')):
           #Nitter上のYouTubeへのリンクをInvidiousのものから直す
           #Nitter上のYouTubeへのリンクをInvidiousのものから直す
776行目: 1,089行目:
           else:
           else:
             url_link = self._invidious_pattern.sub('youtu.be', url_link)
             url_link = self._invidious_pattern.sub('youtu.be', url_link)
           url.replace_with(self._archive_url(url_link)) ##テンプレートArchiveに変化
           url.replace_with(self._archive_url(url_link, accessor)) ##テンプレートArchiveに変化
         elif url.get('href').startswith('https://bibliogram.art/'):
         elif url.get('href').startswith('https://bibliogram.art/'):
           # Nitter上のInstagramへのリンクをBibliogramのものから直す
           # Nitter上のInstagramへのリンクをBibliogramのものから直す
           # Bibliogramは中止されたようなのでそのうちリンクが変わるかも
           # Bibliogramは中止されたようなのでそのうちリンクが変わるかも
           url_link: str = url.get('href').replace('https://bibliogram.art/', 'https://www.instagram.com/')
           url_link: str = url.get('href').replace('https://bibliogram.art/', 'https://www.instagram.com/')
           url.replace_with(self._archive_url(url_link)) ##テンプレートArchiveに変化
           url.replace_with(self._archive_url(url_link, accessor)) ##テンプレートArchiveに変化
         else:
         else:
           url.replace_with(self._archive_url(url.get('href'))) ##テンプレートArchiveに変化
           url.replace_with(self._archive_url(url.get('href'), accessor)) ##テンプレートArchiveに変化
       elif url.text.startswith('@'):
       elif url.text.startswith('@'):
           url_link: str = urljoin(self.TWITTER_URL, url.get('href'))
           url_link: str = urljoin(self.TWITTER_URL, url.get('href'))
           url_text: str = url.text
           url_text: str = url.text
           url.replace_with(self._archive_url(url_link, url_text)) ##テンプレートArchiveに変化
           url.replace_with(self._archive_url(url_link, accessor, url_text)) ##テンプレートArchiveに変化


   def _archive_url(self, url: Final[str], text: Final[str|None] = None) -> str:
   def _archive_url(self, url: Final[str], accessor: AccessorHandler, text: Final[str|None] = None) -> str:
     """URLをArchiveテンプレートでラップする。
     """URLをArchiveテンプレートでラップする。


796行目: 1,109行目:
     Args:
     Args:
       url Final[str]: ラップするURL。
       url Final[str]: ラップするURL。
      accessor AccessorHandler: アクセスハンドラ
       text Final[str|None]: ArchiveテンプレートでURLの代わりに表示する文字列。
       text Final[str|None]: ArchiveテンプレートでURLの代わりに表示する文字列。


804行目: 1,118行目:
       main_url, fragment = url.split('#', maxsplit=1)
       main_url, fragment = url.split('#', maxsplit=1)
       if text is None:
       if text is None:
         return '{{Archive|1=' + unquote(url) + '|2=' + self._archive(main_url) + '#' + fragment + '}}' ##テンプレートArchiveの文字列返す
         return '{{Archive|1=' + unquote(url) + '|2=' + self._archive(main_url, accessor) + '#' + fragment + '}}' ##テンプレートArchiveの文字列返す
       else:
       else:
         return '{{Archive|1=' + unquote(url) + '|2=' + self._archive(main_url) + '#' + fragment + + '|3=' + text + '}}' ##テンプレートArchiveの文字列返す
         return '{{Archive|1=' + unquote(url) + '|2=' + self._archive(main_url, accessor) + '#' + fragment + + '|3=' + text + '}}' ##テンプレートArchiveの文字列返す
     else:
     else:
       if text is None:
       if text is None:
         return '{{Archive|1=' + unquote(url) + '|2=' + self._archive(url) + '}}' ##テンプレートArchiveの文字列返す
         return '{{Archive|1=' + unquote(url) + '|2=' + self._archive(url, accessor) + '}}' ##テンプレートArchiveの文字列返す
       else:
       else:
         return '{{Archive|1=' + unquote(url) + '|2=' + self._archive(url) + '|3=' + text + '}}' ##テンプレートArchiveの文字列返す
         return '{{Archive|1=' + unquote(url) + '|2=' + self._archive(url, accessor) + '|3=' + text + '}}' ##テンプレートArchiveの文字列返す


   def _callinshowlink_url(self, url: Final[str]) -> str:
   def _callinshowlink_url(self, url: Final[str], accessor: AccessorHandler) -> str:
     """URLをCallinShowLinkテンプレートでラップする。
     """URLをCallinShowLinkテンプレートでラップする。


     Args:
     Args:
       url Final[str]: ラップするURL。
       url Final[str]: ラップするURL。
      accessor AccessorHandler: アクセスハンドラ


     Returns:
     Returns:
       str: CallinShowLinkタグでラップしたURL。
       str: CallinShowLinkタグでラップしたURL。
     """
     """
     return '{{CallinShowLink|1=' + url + '|2=' + self._archive(url) + '}}'
     return '{{CallinShowLink|1=' + url + '|2=' + self._archive(url, accessor) + '}}'


   def _archive(self, url: Final[str]) -> str:
   def _archive(self, url: Final[str], accessor: AccessorHandler) -> str:
     """URLから対応するarchive.todayのURLを返す。
     """URLから対応するarchive.todayのURLを返す。


833行目: 1,148行目:
     Args:
     Args:
       url Final[str]: 魚拓を取得するURL。
       url Final[str]: 魚拓を取得するURL。
      accessor AccessorHandler: アクセスハンドラ


     Returns:
     Returns:
838行目: 1,154行目:
     """
     """
     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[str | None] = accessor.request(urljoin(self.ARCHIVE_TODAY, quote(unquote(url), safe='&=+?%'))) ##アクセス用URL使って結果を取得
     if res is None : ##魚拓接続失敗時処理
     if res is None : ##魚拓接続失敗時処理
       print(archive_url + 'にアクセス失敗ナリ。出力されるテキストにはそのまま記載されるナリ。')
       print(archive_url + 'にアクセス失敗ナリ。出力されるテキストにはそのまま記載されるナリ。')
     else:
     else:
       soup: Final[BeautifulSoup] = BeautifulSoup(res.text, 'html.parser') ##beautifulsoupでレスポンス解析
       soup: Final[BeautifulSoup] = BeautifulSoup(res, 'html.parser') ##beautifulsoupでレスポンス解析
       content: bs4.element.Tag = soup.find(id="CONTENT") ##archive.todayの魚拓一覧ページの中身だけ取得
       content: bs4.element.Tag = soup.find(id="CONTENT") ##archive.todayの魚拓一覧ページの中身だけ取得
       if content is None or content.get_text()[:len(self.NO_ARCHIVE)] == self.NO_ARCHIVE: ##魚拓があるかないか判定
       if content is None or content.get_text()[:len(self.NO_ARCHIVE)] == self.NO_ARCHIVE: ##魚拓があるかないか判定
850行目: 1,166行目:
     return archive_url
     return archive_url


   def go_to_new_page(self) -> None | NoReturn:
   def _go_to_new_page(self, accessor: AccessorHandler) -> None | NoReturn:
     """Nitterで次のページに移動する。
     """Nitterで次のページに移動する。


     次のページが無ければプログラムを終了する。
     次のページが無ければプログラムを終了する。
    Args:
      accessor AccessorHandler: アクセスハンドラ
     """
     """
     soup: Final[BeautifulSoup] = BeautifulSoup(self._page.text, 'html.parser') ##beautifulsoupでレスポンス解析
     soup: Final[BeautifulSoup] = BeautifulSoup(self._page, '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")
     new_url: str = '' # ここで定義しないと動かなくなった、FIXME?
     new_url: str = '' # ここで定義しないと動かなくなった、FIXME?
861行目: 1,180行目:
       if show_more.text != self.NEWEST:  ##前ページへのリンクではないか判定
       if show_more.text != self.NEWEST:  ##前ページへのリンクではないか判定
         new_url = urljoin(self.NITTER_INSTANCE, self.CALLINSHOW + '/' + self.TWEETS_OR_REPLIES + show_more.a.get('href')) ##直下のaタグのhrefの中身取ってURL頭部分と合体
         new_url = urljoin(self.NITTER_INSTANCE, self.CALLINSHOW + '/' + self.TWEETS_OR_REPLIES + show_more.a.get('href')) ##直下のaタグのhrefの中身取ってURL頭部分と合体
     res: Final[Response | None] = self._request(new_url) ##接続してHTML取ってくる
     res: Final[str | None] = accessor.request(new_url) ##接続してHTML取ってくる
     if res is None:
     if res is None:
       self._fail()
       self._fail()
     new_page_soup: Final[BeautifulSoup] = BeautifulSoup(res.text, 'html.parser') ##beautifulsoupでレスポンス解析
     new_page_soup: Final[BeautifulSoup] = BeautifulSoup(res, 'html.parser') ##beautifulsoupでレスポンス解析
     if new_page_soup.find(class_="timeline-end") is None: ##ツイートの終端ではtimeline-endだけのページになるので判定
     if new_page_soup.find(class_="timeline-end") is None: ##ツイートの終端ではtimeline-endだけのページになるので判定
       print(res.url + 'に移動しますを')
       print(new_url + 'に移動しますを')
       self._page = res ##まだ残りツイートがあるのでページを返して再度ツイート本文収集
       self._page = res ##まだ残りツイートがあるのでページを返して再度ツイート本文収集
     else:
     else:
       print("急に残りツイートが無くなったな終了するか")
       print("急に残りツイートが無くなったな終了するか")
       self._make_txt()
       self._make_txt()
  def execute(self, krsw: bool=False) -> NoReturn:
    """通信が必要な部分のロジック。
    Args:
      krsw bool: Trueの場合、名前が自動で :const:`~CALLINSHOW` になり、クエリと終わりにするツイートが自動で無しになる。
    """
    # Seleniumドライバーを必ず終了するため、with文を利用する。
    with AccessorHandler() as accessor:
      # 実行前のチェック
      self._check_slash() ##スラッシュが抜けてないかチェック
      self._check_nitter_instance(accessor) ##Nitterが死んでないかチェック
      self._check_archive_instance(accessor) ##archive.todayが死んでないかチェック
      ##Invidiousのインスタンスリストの正規表現パターンを取得
      invidious_url_tuple: Final[tuple[str]] = self._invidious_instances(accessor)
      self._invidious_pattern: Final[re.Pattern] = re.compile('|'.join(invidious_url_tuple))
      # 検索クエリの設定
      self._set_queries(accessor, krsw)
      # ツイートを取得し終えるまでループ
      while True:
        self._get_tweet(accessor)
        self._go_to_new_page(accessor) ##新しいページ取得




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


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


匿名利用者
「https://krsw-wiki.in/wiki/特別:携帯機器差分/94662」から取得