唐澤貴洋Wiki

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

利用者ページ 議論

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

2023年10月1日 (日) 16:40時点における版

3,688 バイト追加 、 2023年10月1日 (日)
→‎コード: v4.0.5 動画を取得できるnitterインスタンスに修正
>Fet-Fe
(→‎コード: v4.0.4 SeleniumAccessor.quit()がself._driverの未定義時にエラーを吐かないよう修正)
>Fet-Fe
(→‎コード: v4.0.5 動画を取得できるnitterインスタンスに修正)
7行目: 7行目:
"""Twitter自動収集スクリプト
"""Twitter自動収集スクリプト


ver4.0.4 2023/9/29恒心
ver4.0.5 2023/10/1恒心


当コードは恒心停止してしまった https://rentry.co/7298g の降臨ショーツイート自動収集スクリプトの復刻改善版です
当コードは恒心停止してしまった https://rentry.co/7298g の降臨ショーツイート自動収集スクリプトの復刻改善版です
32行目: 32行目:
     * Pythonのバージョンは3.11以上
     * Pythonのバージョンは3.11以上
     * 環境は玉葱前提です。
     * 環境は玉葱前提です。
        * Tor Browserを入れておくか、torコマンドでプロキシを立てておくことが必要です。
     * Whonix-Workstation, MacOSで動作確認済
     * Whonix-Workstation, MacOSで動作確認済


39行目: 42行目:
     * requestsも環境によっては入っていないかもしれない
     * requestsも環境によっては入っていないかもしれない


         * $ pip install bs4 requests PySocks selenium
         * ``$ python3 -m pip install bs4 requests PySocks selenium``


     * pipも入っていなければ ``$ sudo apt install pip``
     * pipも入っていなければ ``$ sudo apt install pip``
     * `ffmpeg <https://ffmpeg.org>`_ が入っていると動画も自動取得しますが、無くても動きます
     * `ffmpeg <https://ffmpeg.org>`_ が入っていると動画も自動取得しますが、無くても動きます
     * バグ報告は `利用者・トーク:夜泣き#利用者:夜泣き/スクリプトについて <https://krsw-wiki.org/wiki/利用者・トーク:夜泣き#利用者:夜泣き/スクリプトについて>`_
     * バグ報告は `利用者・トーク:夜泣き#利用者:夜泣き/スクリプトについて \
        <https://krsw-wiki.org/wiki/利用者・トーク:夜泣き#利用者:夜泣き/スクリプトについて>`_
"""
"""


59行目: 63行目:
from collections.abc import Callable
from collections.abc import Callable
from datetime import datetime
from datetime import datetime
from enum import Enum
from logging import Logger, getLogger
from logging import Logger, getLogger
from re import Match
from re import Match
74行目: 79行目:
                                         WebDriverException)
                                         WebDriverException)
from selenium.webdriver.common.by import By
from selenium.webdriver.common.by import By
from selenium.webdriver.firefox.options import Options
from selenium.webdriver.firefox.options import Options as FirefoxOptions
from selenium.webdriver.support import expected_conditions as ec
from selenium.webdriver.support import expected_conditions as ec
from selenium.webdriver.support.wait import WebDriverWait
from selenium.webdriver.support.wait import WebDriverWait
128行目: 133行目:


class RequestsAccessor(AbstractAccessor):
class RequestsAccessor(AbstractAccessor):
     """requestsモジュールでWebサイトに接続するためのクラス。
     """RequestsモジュールでWebサイトに接続するためのクラス。
     """
     """


     HEADERS: Final[dict[str, str]] = {
     HEADERS: Final[dict[str, str]] = {
         'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; rv:91.0) Gecko/20100101 Firefox/91.0'
         'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/118.0'
     }
     }
     """Final[dict[str, str]]: HTTPリクエスト時のヘッダ。
     """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プロキシの設定。
     """
     """


142行目: 154行目:
     }
     }
     """Final[dict[str, str]]: torコマンドを起動しているときのHTTPプロキシの設定。
     """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プロキシの設定。
     """
     """


163行目: 168行目:
     def _execute(self,
     def _execute(self,
                 url: str,
                 url: str,
                 proxies: dict[str,
                 proxies: dict[str, str] | None) -> requests.models.Response:
                              str] | None) -> requests.models.Response:
         """引数のURLにRequestsモジュールでHTTP接続する。
         """引数のURLにrequestsモジュールでHTTP接続する。


         Args:
         Args:
             url str: 接続するURL。
             url (str): 接続するURL。
             proxies dir[str, str]: 接続に利用するプロキシ。
             proxies (dict[str, str] | None): 接続に利用するプロキシ。
            デフォルトでは :func:`~_choose_tor_proxies` で設定した値を利用する。
                デフォルトでは :func:`~_choose_tor_proxies` で設定した値を利用する。
 
        Returns:
            requests.models.Response: レスポンスのオブジェクト。


         Raises:
         Raises:
             requests.exceptions.ConnectionError: Ｔorで接続ができないなど、接続のエラー。
             requests.exceptions.ConnectionError: Ｔorで接続ができないなど、接続のエラー。
             AccessError: ステータスコードが200でない場合のエラー。
             AccessError: ステータスコードが200でない場合のエラー。
        Returns:
            requests.models.Response: レスポンスのオブジェクト。
         """
         """
         sleep(self.WAIT_TIME)  # DoS対策で待つ
         sleep(self.WAIT_TIME)  # DoS対策で待つ
197行目: 201行目:
     def get(self,
     def get(self,
             url: str,
             url: str,
             proxies: dict[str,
             proxies: dict[str, str] | None = None) -> str:
                          str] | None = None) -> str:
         """引数のURLにRequestsモジュールでHTTP接続する。
         """引数のURLにrequestsモジュールでHTTP接続する。


         Args:
         Args:
             url str: 接続するURL。
             url (str): 接続するURL。
             proxies dir[str, str]: 接続に利用するプロキシ。デフォルトでは :func:`~_choose_tor_proxies` で設定した値を利用する。
             proxies (dict[str, str] | None, optional): 接続に利用するプロキシ。
 
                デフォルトでは :func:`~_choose_tor_proxies` で設定した値を利用する。
        Returns:
            str: レスポンスのHTML。


         Raises:
         Raises:
             requests.exceptions.ConnectionError: Ｔorで接続ができないなど、接続のエラー。
             requests.exceptions.ConnectionError: Ｔorで接続ができないなど、接続のエラー。
             AccessError: ステータスコードが200でない場合のエラー。
             AccessError: ステータスコードが200でない場合のエラー。
        Returns:
            str: レスポンスのHTML。
         """
         """
         try:
         try:
221行目: 225行目:


         Args:
         Args:
             url str: 接続するURL
             url (str): 接続するURL。
 
        Returns:
            bytes | None: 画像のバイナリ。画像でなければNone。


         Raises:
         Raises:
             requests.exceptions.ConnectionError: Ｔorで接続ができないなど、接続のエラー。
             requests.exceptions.ConnectionError: Ｔorで接続ができないなど、接続のエラー。
             AccessError: ステータスコードが200でない場合のエラー。
             AccessError: ステータスコードが200でない場合のエラー。
        Returns:
            bytes | None: 画像のバイナリ。画像でなければNone。
         """
         """
         try:
         try:
247行目: 251行目:
         torコマンドのプロキシで接続できるなら :const:`~PROXIES_WITH_COMMAND`。
         torコマンドのプロキシで接続できるなら :const:`~PROXIES_WITH_COMMAND`。
         いずれでもアクセスできなければ異常終了する。
         いずれでもアクセスできなければ異常終了する。
        Raises:
            AccessError: :const:`~TOR_CHECK_URL` にアクセスできない場合のエラー。


         Returns:
         Returns:
             dict[str, str] | None | NoReturn: プロキシ情報。
             dict[str, str] | None | NoReturn: プロキシ情報。
        Raises:
            RuntimeError: :const:`~TOR_CHECK_URL` にアクセスできない場合のエラー。
         """
         """
         logger.info('Ｔorのチェック中ですを')
         logger.info('Ｔorのチェック中ですを')
282行目: 286行目:
                 return self.PROXIES_WITH_COMMAND
                 return self.PROXIES_WITH_COMMAND
             else:
             else:
                 raise RuntimeError('サイトにＴorのIPでアクセスできていないなりを')
                 raise AccessError('サイトにＴorのIPでアクセスできていないなりを')
         except requests.exceptions.ConnectionError as e:
         except requests.exceptions.ConnectionError as e:
             logger.critical(e)
             logger.critical(e)
             logger.critical('通信がＴorのSOCKS proxyを経由していないなりを')
             logger.critical('通信がＴorのSOCKS proxyを経由していないなりを')
             exit(1)
             sys.exit(1)


     @property
     @property
303行目: 307行目:


     TOR_BROWSER_PATHS: MappingProxyType[str, str] = MappingProxyType({
     TOR_BROWSER_PATHS: MappingProxyType[str, str] = MappingProxyType({
         'Windows': '',
         'Windows': 'C:\\Program Files\\Tor Browser\\Browser\\firefox.exe',
         'Darwin': '/Applications/Tor Browser.app/Contents/MacOS/firefox',
         'Darwin': '/Applications/Tor Browser.app/Contents/MacOS/firefox',
         'Linux': '/usr/bin/torbrowser'
         'Linux': '/usr/bin/torbrowser'
     })
     })
     """MappingProxyType[str, str]: OSごとのTor Browserのパス。
     """MappingProxyType[str, str]: OSごとのTor Browserのパス。
    Todo:
        WindowsとLinuxでのTor Browserの実行パスを追加する。
     """
     """


     WAIT_TIME_FOR_INIT: Final[int] = 15
     WAIT_TIME_FOR_INIT: Final[int] = 15
     """Final[int]: 最初のTor接続時の待機時間。
     """Final[int]: 最初のTor接続時の待機時間。
    """
    WAIT_TIME_FOR_RECAPTCHA: Final[int] = 10000
    """Final[int]: reCAPTCHAのための待機時間。
     """
     """


323行目: 328行目:


         Args:
         Args:
             enable_javascript bool: JavaScriptを有効にするかどうか。reCAPTCHA対策に必要。
             enable_javascript (bool): JavaScriptを有効にするかどうか。reCAPTCHA対策に必要。
         """
         """
         self._javascript_enabled: bool = enable_javascript
         self._javascript_enabled: bool = enable_javascript


         options: Options = Options()
         options: FirefoxOptions = FirefoxOptions()
         options.binary_location = self.TOR_BROWSER_PATHS[platform.system()]
         options.binary_location = self.TOR_BROWSER_PATHS[platform.system()]


353行目: 358行目:
             # Torの接続が完了するまで待つ
             # Torの接続が完了するまで待つ
             wait_init.until(ec.url_contains('about:blank'))  # type: ignore
             wait_init.until(ec.url_contains('about:blank'))  # type: ignore
            self.wait: WebDriverWait = WebDriverWait(self._driver,
                                                    self.REQUEST_TIMEOUT)
         except BaseException:
         except BaseException:
             self.quit()
             self.quit()
363行目: 365行目:
         """Seleniumドライバを終了する。
         """Seleniumドライバを終了する。
         """
         """
         if hasattr(self, 'driver'):
         if hasattr(self, '_driver'):
             self._driver.quit()
             self._driver.quit()


384行目: 386行目:
                 WebDriverWait(
                 WebDriverWait(
                     self._driver,
                     self._driver,
                     10000).until(  # type: ignore
                     self.WAIT_TIME_FOR_RECAPTCHA).until(  # type: ignore
                     ec.staleness_of(
                     ec.staleness_of(
                         self._driver.find_element(
                         self._driver.find_element(
402行目: 404行目:


         Args:
         Args:
             url str: 接続するURL。
             url (str): 接続するURL。
 
        Raises:
            AccessError: 通信に失敗した場合のエラー。


         Returns:
         Returns:
437行目: 442行目:


         Args:
         Args:
             use_browser bool: TrueならSeleniumを利用する。FalseならRequestsのみでアクセスする。
             use_browser (bool): TrueならSeleniumを利用する。FalseならRequestsのみでアクセスする。
             enable_javascript bool: SeleniumでJavaScriptを利用する場合はTrue。
             enable_javascript (bool): SeleniumでJavaScriptを利用する場合はTrue。
         """
         """
         self.selenium_accessor: SeleniumAccessor | None = SeleniumAccessor(
         self._selenium_accessor: SeleniumAccessor | None = SeleniumAccessor(
             enable_javascript) if use_browser else None
             enable_javascript) if use_browser else None
         self.requests_accessor: RequestsAccessor = RequestsAccessor()
         self._requests_accessor: RequestsAccessor = RequestsAccessor()


     def __enter__(self) -> Self:
     def __enter__(self) -> Self:
463行目: 468行目:
             traceback (TracebackType | None): コンテキスト内で例外を吐いた場合のトレースバック。
             traceback (TracebackType | None): コンテキスト内で例外を吐いた場合のトレースバック。
         """
         """
         if self.selenium_accessor is not None:
         if self._selenium_accessor is not None:
             self.selenium_accessor.quit()
             self._selenium_accessor.quit()


     def request_once(self, url: str) -> str:
     def request_once(self, url: str) -> str:
470行目: 475行目:


         Args:
         Args:
             url str: 接続するURL。
             url (str): 接続するURL。


         Returns:
         Returns:
482行目: 487行目:
         """
         """
         try:
         try:
             if self.selenium_accessor is not None:
             if self._selenium_accessor is not None:
                 return self.selenium_accessor.get(url)
                 return self._selenium_accessor.get(url)
             else:
             else:
                 return self.requests_accessor.get(url)
                 return self._requests_accessor.get(url)
         except AccessError:
         except AccessError:
             raise
             raise
491行目: 496行目:
     def _request_with_callable(self,
     def _request_with_callable(self,
                               url: str,
                               url: str,
                               request_callable: Callable[[str],
                               request_callable: Callable[[str], Any]
                                                          Any]) -> Any | None:
                              ) -> Any | None:
         """request_callableの実行を :const:`~LIMIT_N_REQUESTS` で定義された回数まで試す。
         """request_callableの実行を :const:`~LIMIT_N_REQUESTS` で定義された回数まで試す。


499行目: 504行目:


         Args:
         Args:
             url str: 接続するURL
             url (str): 接続するURL
             request_callable Callable[[str], Any]: 1回リクエストを行うメソッド。
             request_callable (Callable[[str], Any]): 1回リクエストを行うメソッド。


         Returns:
         Returns:
526行目: 531行目:


         Args:
         Args:
             url str: 接続するURL
             url (str): 接続するURL


         Returns:
         Returns:
537行目: 542行目:


     def request_with_requests_module(self, url: str) -> str | None:
     def request_with_requests_module(self, url: str) -> str | None:
         """requestsモジュールでのHTTP接続を :const:`~LIMIT_N_REQUESTS` で定義された回数まで試す。
         """RequestsモジュールでのHTTP接続を :const:`~LIMIT_N_REQUESTS` で定義された回数まで試す。


         成功すると結果を返す。
         成功すると結果を返す。
543行目: 548行目:


         Args:
         Args:
             url str: 接続するURL
             url (str): 接続するURL


         Returns:
         Returns:
551行目: 556行目:
             失敗かどうかは呼出側で要判定
             失敗かどうかは呼出側で要判定
         """
         """
         return self._request_with_callable(url, self.requests_accessor.get)
         return self._request_with_callable(url, self._requests_accessor.get)


     def request_image(self, url: str) -> bytes | None:
     def request_image(self, url: str) -> bytes | None:
         """requestsモジュールで画像ファイルを取得します。
         """Requestsモジュールで画像ファイルを取得する。


         成功すると結果を返します。
         成功すると結果を返す。
         接続失敗が何度も起きるとNoneを返します。
         接続失敗が何度も起きるとNoneを返す。


         Args:
         Args:
             url Final[str]: 接続するURL
             url (str): 接続するURL


         Returns:
         Returns:
             bytes | None: レスポンスのバイト列。接続失敗が何度も起きるとNoneを返します。
             bytes | None: レスポンスのバイト列。接続失敗が何度も起きるとNoneを返します。
        Note:
            失敗かどうかは呼出側で要判定
         """
         """
         return self._request_with_callable(url,
         return self._request_with_callable(url,
                                           self.requests_accessor.get_image)
                                           self._requests_accessor.get_image)


     @property
     @property
575行目: 583行目:
             dict[str, str] | None: RequestsAccessorオブジェクトのプロキシ設定。
             dict[str, str] | None: RequestsAccessorオブジェクトのプロキシ設定。
         """
         """
         return self.requests_accessor.proxies
         return self._requests_accessor.proxies




class TableBuilder:
class TableBuilder:
    """Wikiの表を組み立てるためのクラス。
    """
     def __init__(self, date: datetime) -> None:
     def __init__(self, date: datetime) -> None:
        """コンストラクタ。
        Args:
            date (datetime): 記録するツイートの最新日付。
        """
         self._tables: list[str] = ['']
         self._tables: list[str] = ['']
         self._count: int = 0  # 記録数
         self._count: int = 0  # 記録数
586行目: 601行目:
     @property
     @property
     def count(self) -> int:
     def count(self) -> int:
        """表に追加したツイートの件数を返す。
        Returns:
            int: 表に追加したツイートの件数。
        """
         return self._count
         return self._count


     def append(self, archived_tweet_url: str, text: str) -> None:
     def append(self, tweet_archive_template: str, text: str) -> None:
         self._tables[0] = '!' + archived_tweet_url + '\n|-\n|\n' \
        """ツイートを表に追加する。
 
        Args:
            archived_tweet_url (str): ツイートのURLをCallinshowLinkテンプレートに入れたもの。
            text (str): ツイートの本文。
        """
         self._tables[-1] = '!' + tweet_archive_template + '\n|-\n|\n' \
             + text \
             + text \
             + '\n|-\n' \
             + '\n|-\n' \
             + self._tables[0] # wikiの文法に変化
             + self._tables[-1]
         self._count += 1
         self._count += 1


     def dump_file(self) -> NoReturn:
     def dump_file(self) -> None:
         """Wikiテーブルをファイル出力し、プログラムを終了する。
         """Wikiテーブルをファイル出力する。
         """
         """
         self._next_day()
         self._next_day()
         result_txt: Final[str] = '\n'.join(self._tables)
         result_txt: Final[str] = '\n'.join(reversed(self._tables))


         with codecs.open('tweet.txt', 'w', 'utf-8') as f:
         with codecs.open('tweet.txt', 'w', 'utf-8') as f:
             f.write(result_txt)
             f.write(result_txt)
         logger.info('テキストファイル手に入ったやで〜')
         logger.info('テキストファイル手に入ったやで〜')
        exit(0)


     def next_day_if_necessary(self, date: datetime | None = None) -> None:
     def next_day_if_necessary(self, date: datetime) -> None:
         if date is None:
         """引数dateがインスタンスの持っている日付より前の場合、日付更新処理をする。
            return
 
         if date.year != self._date.year or date.month != self._date.month or date.day != self._date.day:
        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)
             self._next_day(date)


     def _next_day(self, date: datetime | None = None) -> None:
     def _next_day(self, date: datetime | None = None) -> None:
         if self._tables[0]:
        """Wikiテーブルに日付の見出しを付与する。
             self._tables[0] = self._convert_to_text_table(self._tables[0])
 
             if os.name == 'nt': # Windows
        Args:
                 self._tables[0] = self._date.strftime(
            date (datetime | None, optional): 次に記録するツイートの日付。
                     '\n=== %#m月%#d日 ===\n') + self._tables[0]
        """
         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日のツイートを取得完了ですを'))
                 logger.info(self._date.strftime('%#m月%#d日のツイートを取得完了ですを'))
             else: # Mac or Linux
             else:
                 self._tables[0] = self._date.strftime(
                 self._tables[-1] = self._date.strftime(
                     '\n=== %-m月%-d日 ===\n') + self._tables[0]
                     '\n=== %-m月%-d日 ===\n') + self._tables[-1]
                 logger.info(self._date.strftime('%-m月%-d日のツイートを取得完了ですを'))
                 logger.info(self._date.strftime('%-m月%-d日のツイートを取得完了ですを'))
         if date is not None:
         if date is not None:
            self._tables.insert(0, '')
             self._date = date
             self._date = date
            if self._tables[-1]:
                self._tables.append('')


     def _convert_to_text_table(self, text: str) -> str:
     def _convert_to_text_table(self, text: str) -> str:
         """``self._tables[0]`` にwikiでテーブル表示にするためのヘッダとフッタをつける。
         """Wikiでテーブル表示にするためのヘッダとフッタをつける。


         Args:
         Args:
             text str: ヘッダとフッタがないWikiテーブル。
             text (str): ヘッダとフッタがないWikiテーブル。


         Returns:
         Returns:
643行目: 678行目:


         Args:
         Args:
             text str: ツイートの文字列。
             text (str): ツイートの文字列。


         Returns:
         Returns:
652行目: 687行目:


             Args:
             Args:
                 text str: ツイートの文字列。
                 text (str): ツイートの文字列。


             Returns:
             Returns:
661行目: 696行目:
             while i < len(text):
             while i < len(text):
                 if is_in_archive_template:
                 if is_in_archive_template:
                     if text[i:i + 2] == '}}':
                     if text[i:i + len('}}')] == '}}':
                         is_in_archive_template = False
                         is_in_archive_template = False
                         i += 2
                         i += len('}}')
                 else:
                 else:
                     if (text[i:i + 10] == '{{Archive|'
                     if (text[i:i + len('{{Archive|')] == '{{Archive|'
                             or text[i:i + 10] == '{{archive|'):
                             or text[i:i + len('{{Archive|')] == '{{archive|'):
                         is_in_archive_template = True
                         is_in_archive_template = True
                         i += 10
                         i += len('{{Archive|')
                     elif text[i:i + 8] == 'https://':
                     elif text[i:i + len('https://')] == 'https://':
                         text = text[:i] + \
                         text = text[:i] + \
                             '<nowiki>https://</nowiki>' + text[i + 8:]
                             '<nowiki>https://</nowiki>' + \
                         i += 25
                            text[i + len('https://'):]
                     elif text[i:i + 7] == 'http://':
                         i += len('<nowiki>https://</nowiki>')
                     elif text[i:i + len('http://')] == 'http://':
                         text = text[:i] + \
                         text = text[:i] + \
                             '<nowiki>http://</nowiki>' + text[i + 7:]
                             '<nowiki>http://</nowiki>' + \
                         i += 24
                            text[i + len('http://'):]
                         i += len('<nowiki>http://</nowiki>')
                 i += 1
                 i += 1
             return text
             return text
703行目: 740行目:


         Args:
         Args:
             url str: ラップするURL。
             url (str): ラップするURL。
             archive_url str: ラップするURLの魚拓のURL。
             archive_url (str): ラップするURLの魚拓のURL。
             text str | None: ArchiveテンプレートでURLの代わりに表示する文字列。
             text (str | None, optional): ArchiveテンプレートでURLの代わりに表示する文字列。


         Returns:
         Returns:
721行目: 758行目:


         Args:
         Args:
             url str: ラップするURL。
             url (str): ラップするURL。
             archive_url str: ラップするURLの魚拓のURL。
             archive_url (str): ラップするURLの魚拓のURL。


         Returns:
         Returns:
728行目: 765行目:
         """
         """
         return '{{CallinShowLink|1=' + url + '|2=' + archived_url + '}}'
         return '{{CallinShowLink|1=' + url + '|2=' + archived_url + '}}'
class FfmpegStatus(Enum):
    """ffmpegでの動画保存ステータス。
    """
    MP4 = 1
    """mp4の取得に成功したときのステータス。
    """
    TS = 2
    """tsの取得までは成功したが、mp4への変換に失敗したときのステータス。
    """
    FAILED = 3
    """m3u8からtsの取得に失敗したときのステータス。
    """




737行目: 788行目:
     """
     """


     NITTER_INSTANCE: Final[str] = 'https://nitter.net/'
     NITTER_INSTANCE: Final[str] = 'http://nitter.pjsfkvpxlinjamtawaksbnnaqs2fc2mtvmozrzckxh7f3kis6yea25ad.onion/'
     """Final[str]: Nitterのインスタンス。
     """Final[str]: Nitterのインスタンス。


745行目: 796行目:
         末尾にスラッシュ必須。
         末尾にスラッシュ必須。


    Todo:
        インスタンスによっては画像の取得ができない。
         Tor専用のインスタンスが使えるようになったら変更する。
         Ｔor用のインスタンスでないと動画の取得ができない。
     """
     """


753行目: 804行目:


     ページにアクセスして魚拓があるか調べるのにはonion版のarchive.todayを使用。
     ページにアクセスして魚拓があるか調べるのにはonion版のarchive.todayを使用。
     Note:
     Note:
         末尾にスラッシュ必須。
         末尾にスラッシュ必須。
771行目: 823行目:
     Note:
     Note:
         末尾にスラッシュ必須。
         末尾にスラッシュ必須。
    """
    TWITTER_MEDIA_URL: Final[str] = 'https://pbs.twimg.com/media/'
    """Final[str]: TwitterのメディアのURL。
     """
     """


815行目: 871行目:
         """
         """
         self._check_slash()  # スラッシュが抜けてないかチェック
         self._check_slash()  # スラッシュが抜けてないかチェック
        self._has_ffmpeg: bool = self._check_ffmpeg()  # ffmpegがあるかチェック


     def _set_queries(self, accessor: AccessorHandler, krsw: bool) -> None:
     def _set_queries(self, accessor: AccessorHandler, krsw: bool) -> None:
823行目: 880行目:


         Args:
         Args:
             accessor AccessorHandler: アクセスハンドラ
             accessor (AccessorHandler): アクセスハンドラ
             krsw bool: Trueの場合、名前が自動で :const:`~CALLINSHOW` になり、クエリと終わりにするツイートが自動で無しになる。
             krsw (bool): Trueの場合、名前が :const:`~CALLINSHOW` になり、
                クエリと終わりにするツイートが無しになる。
         """
         """


833行目: 891行目:


         # 検索クエリとページ取得
         # 検索クエリとページ取得
         self.query_strs: list[str] = []
         self._query_strs: list[str] = []
         if krsw:
         if krsw:
             logger.info('クエリは自動的になしにナリます')
             logger.info('クエリは自動的になしにナリます')
842行目: 900行目:
                     + self.TWEETS_OR_REPLIES))
                     + self.TWEETS_OR_REPLIES))
         if page_optional is None:
         if page_optional is None:
             self._fail()
             self._on_fail()
         self._page: str = page_optional
         self._page: str = page_optional


860行目: 918行目:


     def _check_slash(self) -> None | NoReturn:
     def _check_slash(self) -> None | NoReturn:
         """URLの最後にスラッシュが付いていなければエラーを出します。
         """URLの最後にスラッシュが付いていなければエラーを出す。


         Returns:
         Returns:
869行目: 927行目:
         """
         """
         if self.NITTER_INSTANCE[-1] != '/':
         if self.NITTER_INSTANCE[-1] != '/':
             raise RuntimeError('NITTER_INSTANCEの末尾には/が必須です')
             raise RuntimeError('NITTER_INSTANCEの末尾をには/が必須です')
         if self.ARCHIVE_TODAY[-1] != '/':
         if self.ARCHIVE_TODAY[-1] != '/':
             raise RuntimeError('ARCHIVE_TODAYの末尾には/が必須です')
             raise RuntimeError('ARCHIVE_TODAYの末尾をには/が必須です')
         if self.ARCHIVE_TODAY_STANDARD[-1] != '/':
         if self.ARCHIVE_TODAY_STANDARD[-1] != '/':
             raise RuntimeError('ARCHIVE_TODAY_STANDARDの末尾には/が必須です')
             raise RuntimeError('ARCHIVE_TODAY_STANDARDの末尾をには/が必須です')
         if self.TWITTER_URL[-1] != '/':
         if self.TWITTER_URL[-1] != '/':
             raise RuntimeError('TWITTER_URLの末尾には/が必須です')
             raise RuntimeError('TWITTER_URLの末尾をには/が必須です')
 
    def _check_ffmpeg(self) -> bool:
        """ffmpegがインストールされているかチェックする。
 
        Returns:
            bool: ffmpegがインストールされているか。
        """
        return subprocess.run(['which', 'ffmpeg'],
                              stdout=subprocess.DEVNULL,
                              stderr=subprocess.DEVNULL).returncode == 0


     def _check_nitter_instance(
     def _check_nitter_instance(
886行目: 954行目:


         Args:
         Args:
             accessor AccessorHandler: アクセスハンドラ
             accessor (AccessorHandler): アクセスハンドラ


         Returns:
         Returns:
894行目: 962行目:
         try:
         try:
             accessor.request_once(self.NITTER_INSTANCE)
             accessor.request_once(self.NITTER_INSTANCE)
         except AccessError as e: # エラー発生時は終了
         except AccessError as e:
             logger.critical(e)
             logger.critical(e)
             logger.critical('インスタンスが死んでますを')
             logger.critical('インスタンスが死んでますを')
             exit(1)
             sys.exit(1)
         logger.info('Nitter OK')
         logger.info('Nitter OK')


905行目: 973行目:


         Args:
         Args:
             accessor AccessorHandler: アクセスハンドラ
             accessor (AccessorHandler): アクセスハンドラ


         Returns:
         Returns:
916行目: 984行目:
             logger.critical(e)
             logger.critical(e)
             logger.critical('インスタンスが死んでますを')
             logger.critical('インスタンスが死んでますを')
             exit(1)
             sys.exit(1)
         logger.info('archive.today OK')
         logger.info('archive.today OK')


925行目: 993行目:


         Args:
         Args:
             accessor AccessorHandler: アクセスハンドラ
             accessor (AccessorHandler): アクセスハンドラ


         Returns:
         Returns:
             tuple[str, ...] | NoReturn: Invidiousのインスタンスのタプル。Invidiousのインスタンスが死んでいれば終了。
             tuple[str, ...] | NoReturn: Invidiousのインスタンスのタプル。インスタンスが死んでいれば終了。
         """
         """
         logger.info('Invidiousのインスタンスリストを取得中ですを')
         logger.info('Invidiousのインスタンスリストを取得中ですを')
         invidious_json: Final[str] | None = accessor.request_with_requests_module(
         invidious_json: Final[str | None] = (
            'https://api.invidious.io/instances.json')
            accessor.request_with_requests_module(
                'https://api.invidious.io/instances.json')
        )
         if invidious_json is None:
         if invidious_json is None:
             logger.critical('Invidiousが死んでますを')
             logger.critical('Invidiousが死んでますを')
             exit(1)
             sys.exit(1)
         instance_list: list[str] = []
         instance_list: list[str] = []
         for instance_info in json.loads(invidious_json):
         for instance_info in json.loads(invidious_json):
953行目: 1,023行目:


         Args:
         Args:
             accessor AccessorHandler: アクセスハンドラ
             accessor (AccessorHandler): アクセスハンドラ。


         Returns:
         Returns:
971行目: 1,041行目:
                     urljoin(self.NITTER_INSTANCE, account_str))
                     urljoin(self.NITTER_INSTANCE, account_str))
                 if res is None:  # リクエスト失敗判定
                 if res is None:  # リクエスト失敗判定
                     self._fail()
                     self._on_fail()
                 soup: BeautifulSoup = BeautifulSoup(res, 'html.parser')
                 soup: BeautifulSoup = BeautifulSoup(res, 'html.parser')
                 if soup.title == self.NITTER_ERROR_TITLE:
                 if soup.title == self.NITTER_ERROR_TITLE:
981行目: 1,051行目:
     def _get_query(self) -> None:
     def _get_query(self) -> None:
         """検索クエリを標準入力から取得する。
         """検索クエリを標準入力から取得する。
        取得したクエリは ``self.query_strs`` に加えられる。
         """
         """
         logger.info('検索クエリを入れるナリ。複数ある時は一つの塊ごとに入力するナリ。空欄を入れると検索開始ナリ。')
         logger.info('検索クエリを入れるナリ。複数ある時は一つの塊ごとに入力するナリ。空欄を入れると検索開始ナリ。')
989行目: 1,057行目:
         # 空欄が押されるまでユーザー入力受付
         # 空欄が押されるまでユーザー入力受付
         while query_input != '':
         while query_input != '':
             self.query_strs.append(query_input)
             self._query_strs.append(query_input)
             query_input = input()
             query_input = input()
         logger.info('クエリのピースが埋まっていく。')
         logger.info('クエリのピースが埋まっていく。')


     def _fail(self) -> NoReturn:
     def _on_fail(self) -> NoReturn:
         """接続失敗時処理。
         """接続失敗時処理。


1,002行目: 1,070行目:
             logger.critical('取得成功した分だけ発行しますを')
             logger.critical('取得成功した分だけ発行しますを')
             self._table_builder.dump_file()
             self._table_builder.dump_file()
         exit(1)
         sys.exit(1)


     def _stop_word(self) -> str:
     def _stop_word(self) -> str:
1,011行目: 1,079行目:
         """
         """
         logger.info(
         logger.info(
             f'ここにツイートの本文を入れる！すると・・・・なんとそれを含むツイートで記録を中断する！（これは本当の仕様です。）ちなみに、空欄だと{self.LIMIT_N_TWEETS}件まで終了しない。')
             'ここにツイートの本文を入れる！すると・・・・なんとそれを含むツイートで記録を中断する！（これは本当の仕様です。）'
            + f'ちなみに、空欄だと{self.LIMIT_N_TWEETS}件まで終了しない。')
         end_str: Final[str] = input()
         end_str: Final[str] = input()
         return end_str
         return end_str
1,022行目: 1,091行目:


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


         Returns:
         Returns:
1,029行目: 1,099行目:
         """
         """
         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(self.TWITTER_MEDIA_URL, media_name)
         image_bytes: Final[bytes | None] = accessor.request_image(url)
         image_bytes: Final[bytes | None] = accessor.request_image(url)
         if image_bytes is not None:
         if image_bytes is not None:
1,037行目: 1,107行目:
         else:
         else:
             return False
             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での保存ステータス。
        """
        if proxies is not None:
            returncode: int = subprocess.run(
                [
                    'ffmpeg', '-y', '-http_proxy',
                    'proxies["http"]', '-i',
                    urljoin(self.NITTER_INSTANCE, media_path),
                    '-c', 'copy', ts_filename
                ],
                stdout=subprocess.DEVNULL).returncode
        else:
            returncode: int = subprocess.run(
                [
                    'ffmpeg', '-y', '-i',
                    urljoin(self.NITTER_INSTANCE, media_path),
                    '-c', 'copy', ts_filename
                ],
                stdout=subprocess.DEVNULL).returncode
        # 取得成功したらtsをmp4に変換
        if returncode == 0:
            ts2mp4_returncode: 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:
     def _tweet_date(self, tweet: Tag) -> datetime:
1,042行目: 1,161行目:


         Args:
         Args:
             tweet Tag: ツイート内容である ``.timeline-item`` タグを表すbeautifulsoup4タグ。
             tweet (Tag): ツイート内容である ``.timeline-item`` タグを表すbeautifulsoup4タグ。


         Returns:
         Returns:
1,067行目: 1,186行目:


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


         Returns:
         Returns:
             str: Wiki記法でのファイルへのリンクの文字列。
             str: Wiki記法でのファイルへのリンクの文字列。
         """
         """
        # 引用リツイート内のメディアを選択しないように.tweet-body直下の.attachmentsを選択
         tweet_media: Tag | None = tweet.select_one(
         tweet_media: Tag | None = tweet.select_one(
             '.tweet-body > .attachments') # 引用リツイート内のメディアを選択しないように.tweet-body直下の.attachmentsを選択
             '.tweet-body > .attachments')
         media_txt: str = ''
         media_txt: str = ''
         if tweet_media is not None:
         if tweet_media is not None:
1,083行目: 1,203行目:
                     href: str | list[str] | None = image_a.get('href')
                     href: str | list[str] | None = image_a.get('href')
                     assert isinstance(href, str)
                     assert isinstance(href, str)
                    matched: Match[str] | None = re.search(
                        r'%2F([^%]*\.jpg)|%2F([^%]*\.jpeg)|%2F([^%]*\.png)|%2F([^%]*\.gif)',
                        href)
                    assert matched is not None
                     media_name: str = [
                     media_name: str = [
                         group for group in re.search(
                         group for group in matched.groups() if group][0]
                            r'%2F([^%]*\.jpg)|%2F([^%]*\.jpeg)|%2F([^%]*\.png)|%2F([^%]*\.gif)',
                            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, accessor):
                     if self._download_media(media_name, accessor):
                         logger.info(
                         logger.info(
                             os.path.join(
                             os.path.join(self.MEDIA_DIR, media_name)
                                self.MEDIA_DIR,
                                media_name)
                             + ' をアップロードしなければない。')
                             + ' をアップロードしなければない。')
                     else:
                     else:
                         logger.info(
                         logger.info(
                             urljoin(
                             urljoin(self.TWITTER_MEDIA_URL, media_name)
                                'https://pbs.twimg.com/media/',
                                media_name)
                             + ' をアップロードしなければない。')
                             + ' をアップロードしなければない。')
                 except AttributeError:
                 except AttributeError:
1,111行目: 1,229行目:
                     logger.error(f'{tweet_url}の画像が取得できませんでしたを 当職無能')
                     logger.error(f'{tweet_url}の画像が取得できませんでしたを 当職無能')
                     media_list.append('[[ファイル:(画像の取得ができませんでした)|240px]]')
                     media_list.append('[[ファイル:(画像の取得ができませんでした)|240px]]')
             # ツイートの動画の取得。ffmpegが入っていなければ手動で取得すること
             # ツイートの動画の取得。ffmpegが入っていなければ手動で取得すること
             for i, video_container in enumerate(
             for i, video_container in enumerate(
1,122行目: 1,241行目:
                     self.TWITTER_URL,
                     self.TWITTER_URL,
                     re.sub('#[^#]*$', '', href))  # ツイートのURL作成
                     re.sub('#[^#]*$', '', href))  # ツイートのURL作成
                if not self._has_ffmpeg:
                    logger.error(f'ffmpegがないため{tweet_url}の動画が取得できませんでしたを')
                    media_list.append('[[ファイル:(動画の取得ができませんでした)|240px]]')
                    continue
                # videoタグがない場合は取得できない
                 video = video_container.select_one('video')
                 video = video_container.select_one('video')
                 if video is None:
                 if video is None:
1,128行目: 1,254行目:
                     continue
                     continue


                 if subprocess.run(['which',
                 data_url: str | list[str] | None = video.get('data-url')
                                  'ffmpeg'],
                assert isinstance(data_url, str)
                                  stdout=subprocess.DEVNULL,
                matched: Match[str] | None = re.search(r'[^\/]+$', data_url)
                                  stderr=subprocess.DEVNULL).returncode != 0:
                assert matched is not None
                    logger.error(f'ffmpegがないため{tweet_url}の動画が取得できませんでしたを')
                media_path: str = unquote(matched.group(0))
                    media_list.append('[[ファイル:(動画の取得ができませんでした)|240px]]')
                tweet_id: str = tweet_url.split('/')[-1]
                else:  # ffmpegがある場合
                ts_filename: str = (
                    # TODO: ブロックが大きすぎるので別メソッドに切り出す
                    f'{os.path.join(self.MEDIA_DIR, tweet_id)}_{i}.ts'
                    data_url: str | list[str] | None = video.get('data-url')
                )
                    assert isinstance(data_url, str)
                mp4_filename: str = (
                    matched: Match[str] | None = re.search(
                    f'{os.path.join(self.MEDIA_DIR, tweet_id)}_{i}.mp4'
                        r'[^\/]+$', data_url)
                )
                    assert matched is not None
 
                    media_url: str = unquote(matched.group(0))
                match self._download_m3u8(
                    tweet_id: str = tweet_url.split('/')[-1]
                        media_path,
                    # 動画のダウンロード
                        ts_filename,
                    if accessor.proxies is not None:
                        mp4_filename,
                        returncode: int = subprocess.run(
                        accessor.proxies):
                            [
                    case FfmpegStatus.MP4:
                                'ffmpeg', '-y', '-http_proxy',
                        logger.info(f'{mp4_filename}をアップロードしなければない。')
                                'accessor.proxies["http"]', '-i',
                         media_list.append(f'[[ファイル:{tweet_id}_{i}.mp4|240px]]')
                                urljoin(self.NITTER_INSTANCE, media_url),
                    case FfmpegStatus.TS:
                                '-c', 'copy',
                        logger.info(f'{ts_filename}.tsをmp4に変換してアップロードしなければない。')
                                f'{os.path.join(self.MEDIA_DIR, tweet_id)}_{i}.ts'
                            ],
                            stdout=subprocess.DEVNULL).returncode
                    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
                    # 取得成功したらtsをmp4に変換
                    if returncode == 0:
                        ts2mp4_returncode: int = subprocess.run(
                            [
                                'ffmpeg', '-y', '-i',
                                f'{os.path.join(self.MEDIA_DIR, tweet_id)}_{i}.ts',
                                '-acodec', 'copy', '-vcodec', 'copy',
                                f'{os.path.join(self.MEDIA_DIR, tweet_id)}_{i}.mp4'
                            ],
                            stdout=subprocess.DEVNULL).returncode
                         if ts2mp4_returncode == 0:
                            logger.info(
                                f'{os.path.join(self.MEDIA_DIR, tweet_id)}_{i}.mp4をアップロードしなければない。')
                        else:
                            logger.info(
                                f'{os.path.join(self.MEDIA_DIR, tweet_id)}_{i}.tsをmp4に変換してアップロードしなければない。')
                         media_list.append(f'[[ファイル:{tweet_id}_{i}.mp4|240px]]')
                         media_list.append(f'[[ファイル:{tweet_id}_{i}.mp4|240px]]')
                     else:
                     case _:
                         logger.error(f'{tweet_url}の動画が取得できませんでしたを 当職無能')
                         logger.error(f'{tweet_url}の動画が取得できませんでしたを 当職無能')
                         media_list.append('[[ファイル:(動画の取得ができませんでした)|240px]]')
                         media_list.append('[[ファイル:(動画の取得ができませんでした)|240px]]')
             media_txt = ' '.join(media_list)
             media_txt = ' '.join(media_list)
         return media_txt
         return media_txt
1,193行目: 1,292行目:


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


         Returns:
         Returns:
1,220行目: 1,319行目:


         Args:
         Args:
             tweet Tag: ツイート内容である ``.timeline-item`` タグを表すbeautifulsoup4タグ。
             tweet (Tag): ツイート内容である ``.timeline-item`` タグを表すbeautifulsoup4タグ。


         Returns:
         Returns:
1,242行目: 1,341行目:
                 assert poll_choice_option is not None
                 assert poll_choice_option is not None
                 if 'leader' in poll_meter['class']:
                 if 'leader' in poll_meter['class']:
                     poll_txt += f'<br>\n&nbsp; <span style="display: inline-block; width: 30em; background: linear-gradient(to right, rgba(29, 155, 240, 0.58) 0 {ratio}, transparent {ratio} 100%); font-weight: bold;">' + \
                     poll_txt += ('<br>\n'
                                '&nbsp; <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>'
                         ratio + ' ' + poll_choice_option.text + '</span>'
                 else:
                 else:
                     poll_txt += f'<br>\n&nbsp; <span style="display: inline-block; width: 30em; background: linear-gradient(to right, rgb(207, 217, 222) 0 {ratio}, transparent {ratio} 100%);">' + \
                     poll_txt += ('<br>\n'
                                '&nbsp; <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>'
                         ratio + ' ' + poll_choice_option.text + '</span>'
             poll_txt += '<br>\n&nbsp; <span style="font-size: small;">' + \
             poll_txt += '<br>\n&nbsp; <span style="font-size: small;">' + \
1,258行目: 1,368行目:


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


         Returns:
         Returns:
1,275行目: 1,385行目:
                 timeline_item_list.append(item_or_list)
                 timeline_item_list.append(item_or_list)
         return timeline_item_list
         return timeline_item_list
    def _get_tweet(self, accessor: AccessorHandler) -> None | NoReturn:
        """ページからツイート本文を ``TableBuilder`` インスタンスに収めていく。
        ツイートの記録を中断するための文を発見するか、記録したツイート数が :const:`~LIMIT_N_TWEETS` 件に達したら終了する。
        Args:
            accessor AccessorHandler: アクセスハンドラ
        """
        soup: Final[BeautifulSoup] = BeautifulSoup(
            self._page, 'html.parser')  # beautifulsoupでレスポンス解析
        tweets: Final[list[Tag]] = self._get_timeline_items(
            soup)  # 一ツイートのブロックごとにリストで取得
        for tweet in tweets:  # 一ツイート毎に処理
            tweet_a: 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:
                # クエリが指定されている場合、一つでも含まないツイートは処理しない、TODO: 未テスト
                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
            tweet_link: Tag | NavigableString | None = tweet.find(
                class_='tweet-link')
            assert isinstance(tweet_link, Tag)
            href: str | list[str] | None = tweet_link.get('href')
            assert isinstance(href, str)
            tweet_url: str = urljoin(
                self.TWITTER_URL,
                re.sub('#[^#]*$', '', href))
            date: datetime = self._tweet_date(tweet)
            self._table_builder.next_day_if_necessary(date)
            archived_tweet_url: str = self._callinshowlink_url(
                tweet_url, accessor)
            tweet_content: Tag | NavigableString | None = tweet.find(
                class_='tweet-content media-body')
            assert isinstance(tweet_content, Tag)
            self._archive_soup(tweet_content, accessor)
            media_txt: str = self._get_tweet_media(tweet, accessor)
            quote_txt: str = self._get_tweet_quote(tweet, accessor)
            poll_txt: str = self._get_tweet_poll(tweet)
            self._table_builder.append(
                archived_tweet_url, '<br>\n'.join(
                    filter(
                        None, [
                            self._table_builder.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()
            if self._table_builder.count >= self.LIMIT_N_TWEETS:
                logger.info(f'{self.LIMIT_N_TWEETS}件も記録している。もうやめにしませんか。')
                self._table_builder.dump_file()


     def _archive_soup(
     def _archive_soup(
1,355行目: 1,395行目:


         Args:
         Args:
             tag Tag: ツイート内容の本文である ``.media-body`` タグを表すbeautifulsoup4タグ。
             tag (Tag): ツイート内容の本文である ``.media-body`` タグを表すbeautifulsoup4タグ。
             accessor AccessorHandler: アクセスハンドラ
             accessor (AccessorHandler): アクセスハンドラ
         """
         """
         urls_in_tweet: Final[ResultSet[Tag]] = tag.find_all(
         urls_in_tweet: Final[ResultSet[Tag]] = tag.find_all(
1,415行目: 1,455行目:


         Args:
         Args:
             url str: ラップするURL。
             url (str): ラップするURL。
             accessor AccessorHandler: アクセスハンドラ。
             accessor (AccessorHandler): アクセスハンドラ。
             text str|None: ArchiveテンプレートでURLの代わりに表示する文字列。
             text (str | None, optional): ArchiveテンプレートでURLの代わりに表示する文字列。


         Returns:
         Returns:
1,434行目: 1,474行目:


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


         Returns:
         Returns:
1,451行目: 1,491行目:


         Args:
         Args:
             url str: 魚拓を取得するURL。
             url (str): 魚拓を取得するURL。
             accessor AccessorHandler: アクセスハンドラ
             accessor (AccessorHandler): アクセスハンドラ。


         Returns:
         Returns:
1,461行目: 1,501行目:
             quote(
             quote(
                 unquote(url),
                 unquote(url),
                 safe='&=+?%')) # wikiに載せるとき用URLで失敗するとこのままhttps://archive.ph/https%3A%2F%2Fxxxxxxxxの形で返される
                 safe='&=+?%'))
         res: Final[str | None] = accessor.request(urljoin(
         res: Final[str | None] = accessor.request(urljoin(
             self.ARCHIVE_TODAY, quote(unquote(url), safe='&=+?%')))
             self.ARCHIVE_TODAY, quote(unquote(url), safe='&=+?%')))
         if res is None:  # 魚拓接続失敗時処理
         if res is None:  # 魚拓接続失敗時処理
            # https://archive.ph/https%3A%2F%2Fxxxxxxxxの形で返される
             logger.error(archive_url + 'にアクセス失敗ナリ。出力されるテキストにはそのまま記載されるナリ。')
             logger.error(archive_url + 'にアクセス失敗ナリ。出力されるテキストにはそのまま記載されるナリ。')
         else:
         else:
1,470行目: 1,511行目:
             content: Tag | NavigableString | None = soup.find(
             content: Tag | NavigableString | None = soup.find(
                 id='CONTENT')  # archive.todayの魚拓一覧ページの中身だけ取得
                 id='CONTENT')  # archive.todayの魚拓一覧ページの中身だけ取得
             if content is None or content.get_text()[:len(
             if (content is None or content.get_text()[:len(self.NO_ARCHIVE)]
                    self.NO_ARCHIVE)] == self.NO_ARCHIVE:  # 魚拓があるかないか判定
                    == self.NO_ARCHIVE):  # 魚拓があるかないか判定
                 logger.warning(url + 'の魚拓がない。これはいけない。')
                 logger.warning(url + 'の魚拓がない。これはいけない。')
             else:
             else:
1,482行目: 1,523行目:
                     self.ARCHIVE_TODAY, self.ARCHIVE_TODAY_STANDARD)
                     self.ARCHIVE_TODAY, self.ARCHIVE_TODAY_STANDARD)
         return archive_url
         return archive_url
    def _get_tweet(self, accessor: AccessorHandler) -> None | NoReturn:
        """ページからツイート本文を ``TableBuilder`` インスタンスに収めていく。
        ツイートの記録を中断するための文を発見するか、記録したツイート数が :const:`~LIMIT_N_TWEETS` 件に達したら終了する。
        Args:
            accessor (AccessorHandler): アクセスハンドラ
        Returns:
            None | NoReturn: 終わりにするツイートを発見するか、記録件数が上限に達したら終了。
        """
        soup: Final[BeautifulSoup] = BeautifulSoup(
            self._page, 'html.parser')
        tweets: Final[list[Tag]] = self._get_timeline_items(
            soup)  # 一ツイートのブロックごとにリストで取得
        for tweet in tweets:  # 一ツイート毎に処理
            tweet_a: 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
            tweet_link: Tag | NavigableString | None = tweet.find(
                class_='tweet-link')
            assert isinstance(tweet_link, Tag)
            href: str | list[str] | None = tweet_link.get('href')
            assert isinstance(href, str)
            tweet_url: str = urljoin(
                self.TWITTER_URL,
                re.sub('#[^#]*$', '', href))
            date: datetime = self._tweet_date(tweet)
            self._table_builder.next_day_if_necessary(date)
            archived_tweet_url: str = self._callinshowlink_url(
                tweet_url, accessor)
            tweet_content: Tag | NavigableString | None = tweet.find(
                class_='tweet-content media-body')
            assert isinstance(tweet_content, Tag)
            self._archive_soup(tweet_content, accessor)
            media_txt: str = self._get_tweet_media(tweet, accessor)
            quote_txt: str = self._get_tweet_quote(tweet, accessor)
            poll_txt: str = self._get_tweet_poll(tweet)
            self._table_builder.append(
                archived_tweet_url, '<br>\n'.join(
                    filter(
                        None,
                        [
                            self._table_builder.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()
                sys.exit(0)
            if self._table_builder.count >= self.LIMIT_N_TWEETS:
                logger.info(f'{self.LIMIT_N_TWEETS}件も記録している。もうやめにしませんか。')
                self._table_builder.dump_file()
                sys.exit(0)


     def _go_to_new_page(self, accessor: AccessorHandler) -> None | NoReturn:
     def _go_to_new_page(self, accessor: AccessorHandler) -> None | NoReturn:
1,489行目: 1,608行目:


         Args:
         Args:
             accessor AccessorHandler: アクセスハンドラ
             accessor (AccessorHandler): アクセスハンドラ
 
        Returns:
            None | NoReturn: Nitterで次のページが無ければ終了。
         """
         """
         soup: Final[BeautifulSoup] = BeautifulSoup(self._page, 'html.parser')
         soup: Final[BeautifulSoup] = BeautifulSoup(self._page, 'html.parser')
         show_mores: Final[ResultSet[Tag]
         show_mores: Final[ResultSet[Tag]] = soup.find_all(class_='show-more')
                          ] = soup.find_all(class_='show-more')
         new_url: str = ''
         new_url: str = '' # ここで定義しないと動かなくなった、FIXME?
         for show_more in show_mores:  # show-moreに次ページへのリンクか前ページへのリンクがある
         for show_more in show_mores:  # show-moreに次ページへのリンクか前ページへのリンクがある
             if show_more.text != self.NEWEST:  # 前ページへのリンクではないか判定
             if show_more.text != self.NEWEST:  # 前ページへのリンクではないか判定
1,509行目: 1,630行目:
         res: Final[str | None] = accessor.request(new_url)
         res: Final[str | None] = accessor.request(new_url)
         if res is None:
         if res is None:
             self._fail()
             self._on_fail()
         new_page_soup: Final[BeautifulSoup] = BeautifulSoup(res, 'html.parser')
         new_page_soup: Final[BeautifulSoup] = BeautifulSoup(res, 'html.parser')
         if new_page_soup.find(
         if new_page_soup.find(
1,519行目: 1,640行目:
             logger.info('急に残りツイートが無くなったな終了するか')
             logger.info('急に残りツイートが無くなったな終了するか')
             self._table_builder.dump_file()
             self._table_builder.dump_file()
            sys.exit(0)


     def execute(self, krsw: bool = False, use_browser: bool = True,
     def execute(self, krsw: bool = False, use_browser: bool = True,
1,525行目: 1,647行目:


         Args:
         Args:
             krsw bool: Trueの場合、名前が自動で :const:`~CALLINSHOW` になり、クエリと終わりにするツイートが自動で無しになる。
             krsw (bool, optional): Trueの場合、名前が自動で :const:`~CALLINSHOW` になり、
             use_browser bool: TrueならSeleniumを利用する。FalseならRequestsのみでアクセスする。
                クエリと終わりにするツイートが自動で無しになる。
             enable_javascript bool: SeleniumでJavaScriptを利用する場合はTrue。
             use_browser (bool, optional): TrueならSeleniumを利用する。
                FalseならRequestsのみでアクセスする。
             enable_javascript (bool, optional): SeleniumでJavaScriptを利用する場合は
                True。
         """
         """
         # Seleniumドライバーを必ず終了するため、with文を利用する。
         # Seleniumドライバーを必ず終了するため、with文を利用する。
1,535行目: 1,660行目:
             self._check_archive_instance(accessor)
             self._check_archive_instance(accessor)
             # Invidiousのインスタンスリストの正規表現パターンを取得
             # Invidiousのインスタンスリストの正規表現パターンを取得
             invidious_url_tuple: Final[tuple[str, ...]
             invidious_url_tuple: Final[tuple[str, ...]] = (
                                      ] = self._invidious_instances(accessor)
                self._invidious_instances(accessor)
            )
             self._invidious_pattern: re.Pattern[str] = re.compile(
             self._invidious_pattern: re.Pattern[str] = re.compile(
                 '|'.join(invidious_url_tuple))
                 '|'.join(invidious_url_tuple))
1,553行目: 1,679行目:
             sys.version_info.major == 3 and sys.version_info.minor < 11):
             sys.version_info.major == 3 and sys.version_info.minor < 11):
         logger.critical('Pythonのバージョンを3.11以上に上げて下さい')
         logger.critical('Pythonのバージョンを3.11以上に上げて下さい')
         exit(1)
         sys.exit(1)
     parser: ArgumentParser = ArgumentParser()
     parser: ArgumentParser = ArgumentParser()
     parser.add_argument(
     parser.add_argument(
匿名利用者
「https://krsw-wiki.in/wiki/特別:携帯機器差分/94667」から取得