マヨケーがポアされたため、現在はロシケーがメインとなっています。

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

ナビゲーションに移動 検索に移動

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

2024年11月9日 (土) 16:59時点における版

1 バイト除去 、 2024年11月9日 (土)
→‎コード: v4.4.2 WikiのURLをkrsw-wiki.inに変更
>Fet-Fe
(→‎コード: v4.4.0 -u --krswを指定した時にも上限以内で引っかかった書き込みがあると自動停止する機能を追加)
(→‎コード: v4.4.2 WikiのURLをkrsw-wiki.inに変更)
 
11行目: 11行目:
"""Twitter自動収集スクリプト
"""Twitter自動収集スクリプト


ver4.4.0 2024/9/4恒心
ver4.4.2 2024/11/9恒心


当コードは恒心停止してしまった https://rentry.co/7298g の降臨ショーツイート自動収集スクリプトの復刻改善版です。
当コードは恒心停止してしまった https://rentry.co/7298g の降臨ショーツイート自動収集スクリプトの復刻改善版です。
19行目: 19行目:
Examples:
Examples:
     定数類は状況に応じて変えてください。:class:`~UserProperties` で変更できます。
     定数類は状況に応じて変えてください。:class:`~UserProperties` で変更できます。
     ::
     ::


24行目: 25行目:


     オプションに ``--krsw`` とつけると自動モードになります。
     オプションに ``--krsw`` とつけると自動モードになります。
     ::
     ::


33行目: 35行目:
     ``--no-browser`` オプションでTor Browserを使用しないモードに、
     ``--no-browser`` オプションでTor Browserを使用しないモードに、
     ``--disable-script`` オプションでTor BrowserのJavaScriptを使用しないモードになります。
     ``--disable-script`` オプションでTor BrowserのJavaScriptを使用しないモードになります。
     ``--search-unarchived`` オプションでは、従来の通りNitterからツイートを収集するモードになります（廃止予定）。
     ``--search-unarchived`` オプションでは、従来の通りNitterからツイートを収集するモードになります (廃止予定)。


Note:
Note:
49行目: 51行目:
     * requests, typing_extensionsも環境によっては入っていないかもしれない
     * requests, typing_extensionsも環境によっては入っていないかもしれない


         * ``$ python3 -m pip install bs4 requests PySocks selenium typing_extensions``
    .. code-block:: bash
 
         $ python3 -m pip install bs4 requests PySocks selenium typing_extensions


     * 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.in/wiki/利用者・トーク:夜泣き#利用者:夜泣き/スクリプトについて>`_
"""
"""


import codecs
import json
import json
import logging
import logging
76行目: 79行目:
from datetime import datetime
from datetime import datetime
from enum import Enum
from enum import Enum
from pathlib import Path
from time import sleep
from time import sleep
from traceback import TracebackException
from traceback import TracebackException
354行目: 358行目:
             else:
             else:
                 raise AccessError('サイトにＴorのIPでアクセスできていないなりを')
                 raise AccessError('サイトにＴorのIPでアクセスできていないなりを')
         except requests.exceptions.ConnectionError as e:
         except requests.exceptions.ConnectionError:
            logger.critical(e)
             logger.critical('通信がＴorのSOCKS proxyを経由していないなりを', exc_info=True)
             logger.critical('通信がＴorのSOCKS proxyを経由していないなりを')
             sys.exit(1)
             sys.exit(1)


566行目: 569行目:


     def _request_with_callable[T](
     def _request_with_callable[T](
            self,
        self,
            url: str,
        url: str,
            request_callable: Callable[[str], T]) -> T | None:
        request_callable: Callable[[str], T]
    ) -> T | None:
         """`request_callable` の実行を :const:`~LIMIT_N_REQUESTS` で定義された回数まで試す。
         """`request_callable` の実行を :const:`~LIMIT_N_REQUESTS` で定義された回数まで試す。


587行目: 591行目:
         for i in range(self.LIMIT_N_REQUESTS):
         for i in range(self.LIMIT_N_REQUESTS):
             try:
             try:
                 res: Final[T] = request_callable(url)
                 res: T = request_callable(url)
             except AccessError:
             except AccessError:
                 logger.warning(
                 logger.warning(
                     url + 'への通信失敗ナリ  '
                     url + 'への通信失敗ナリ  '
                     f'{i + 1}/{self.LIMIT_N_REQUESTS}回')
                     f'{i + 1}/{self.LIMIT_N_REQUESTS}回')
                logger.debug('エラーログ', exc_info=True)
                 if i < self.LIMIT_N_REQUESTS:
                 if i < self.LIMIT_N_REQUESTS:
                     sleep(self.WAIT_TIME_FOR_ERROR)  # 失敗時は長めに待つ
                     sleep(self.WAIT_TIME_FOR_ERROR)  # 失敗時は長めに待つ
677行目: 682行目:
     Args:
     Args:
         date (datetime | None, optional): 記録するツイートの最新日付。デフォルトは今日の日付。
         date (datetime | None, optional): 記録するツイートの最新日付。デフォルトは今日の日付。
    Attributes:
        _tables (list[str]): ツイートのリストを日毎にまとめたもの。\
            一番最後の要素が `_date` に対応し、最初の要素が最近の日付となる。
        _count (int): 表に追加したツイートの件数。
        _date (datetime): 現在収集中のツイートの日付。
     """
     """
     FILENAME: Final[str] = UserProperties.filename
     FILENAME: Final[str] = UserProperties.filename
708行目: 719行目:
         """Wikiテーブルをファイル出力する。"""
         """Wikiテーブルをファイル出力する。"""
         self._next_day()
         self._next_day()
         result_txt: Final[str] = '\n'.join(reversed(self._tables))
         Path(self.FILENAME).write_text(
 
            '\n'.join(reversed(self._tables)), 'utf-8'
        with codecs.open(self.FILENAME, 'w', 'utf-8') as f:
        )
            f.write(result_txt)
         logger.info('テキストファイル手に入ったやで〜')
         logger.info('テキストファイル手に入ったやで〜')


826行目: 836行目:
     @staticmethod
     @staticmethod
     def archive_url(
     def archive_url(
            url: str,
        url: str,
            archived_url: str,
        archived_url: str,
            text: str | None = None) -> str:
        text: str | None = None
    ) -> str:
         """URLをArchiveテンプレートでラップする。
         """URLをArchiveテンプレートでラップする。


984行目: 995行目:


     def _set_queries(
     def _set_queries(
            self, accessor: AccessorHandler, krsw: str | None) -> bool:
        self, accessor: AccessorHandler, krsw: str | None
    ) -> bool:
         """検索条件を設定する。
         """検索条件を設定する。


1,102行目: 1,114行目:
         try:
         try:
             accessor.request_once(self.NITTER_INSTANCE)
             accessor.request_once(self.NITTER_INSTANCE)
         except AccessError as e:
         except AccessError:
            logger.critical(e)
             logger.critical('インスタンスが死んでますを', exc_info=True)
             logger.critical('インスタンスが死んでますを')
             sys.exit(1)
             sys.exit(1)
         logger.info('Nitter OK')
         logger.info('Nitter OK')
1,120行目: 1,131行目:
         try:
         try:
             accessor.request_once(self.ARCHIVE_TODAY)
             accessor.request_once(self.ARCHIVE_TODAY)
         except AccessError as e:  # エラー発生時は終了
         except AccessError:  # エラー発生時は終了
            logger.critical(e)
             logger.critical('インスタンスが死んでますを', exc_info=True)
             logger.critical('インスタンスが死んでますを')
             sys.exit(1)
             sys.exit(1)
         logger.info('archive.today OK')
         logger.info('archive.today OK')


     def _invidious_instances(
     def _invidious_instances(
            self, accessor: AccessorHandler) -> tuple[str, ...]:
        self, accessor: AccessorHandler
    ) -> tuple[str, ...]:
         """Invidiousのインスタンスのタプルを取得する。
         """Invidiousのインスタンスのタプルを取得する。


1,171行目: 1,182行目:
                 + 'になりますを')
                 + 'になりますを')
             print('> ', end='')
             print('> ', end='')
             account_str: Final[str] = input()
             account_str: str = input()
             # 空欄で降臨ショー
             # 空欄で降臨ショー
             if account_str == '':
             if account_str == '':
                 return self.CALLINSHOW
                 return self.CALLINSHOW
             else:
             else:
                 res: Final[str | None] = accessor.request(
                 res: str | None = accessor.request(
                     urljoin(self.NITTER_INSTANCE, account_str))
                     urljoin(self.NITTER_INSTANCE, account_str))
                 if res is None:  # リクエスト失敗判定
                 if res is None:  # リクエスト失敗判定
                     self._on_fail(accessor)
                     self._on_fail(accessor)
                     return None
                     return None
                 soup: Final[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:
                     print(account_str + 'は実在の人物ではありませんでした')
                     print(account_str + 'は実在の人物ではありませんでした')
1,232行目: 1,243行目:


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


1,250行目: 1,262行目:
         image_bytes: Final[bytes | None] = accessor.request_image(media_url)
         image_bytes: Final[bytes | None] = accessor.request_image(media_url)
         if image_bytes is not None:
         if image_bytes is not None:
             with open(os.path.join(self.MEDIA_DIR, media_name), 'wb') as f:
             Path(self.MEDIA_DIR, media_name).write_bytes(image_bytes)
                f.write(image_bytes)
             return True
             return True
         else:
         else:
1,257行目: 1,268行目:


     def _download_m3u8(
     def _download_m3u8(
            self,
        self,
            media_url: str,
        media_url: str,
            ts_filename: str | None,
        ts_filename: str | None,
            mp4_filename: str,
        mp4_filename: str,
            proxies: dict[str, str] | None) -> FfmpegStatus:
        proxies: dict[str, str] | None
    ) -> FfmpegStatus:
         """ffmpegで動画をダウンロードし、:const:`~MEDIA_DIR` に保存する。
         """ffmpegで動画をダウンロードし、:const:`~MEDIA_DIR` に保存する。


1,340行目: 1,352行目:


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


1,363行目: 1,376行目:
             for image_a in tweet_media.select('.attachment.image a'):
             for image_a in tweet_media.select('.attachment.image a'):
                 try:
                 try:
                     href: Final[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)
                     img_matched: Final[re.Match[str] | None] = (
                     img_matched: re.Match[str] | None = (
                         self._img_ext_pattern.search(href))
                         self._img_ext_pattern.search(href))
                     assert img_matched is not None
                     assert img_matched is not None
                     media_name: Final[str] = img_matched.group(1)
                     media_name: str = img_matched.group(1)
                     media_list.append(f'[[ファイル:{media_name}|240px]]')
                     media_list.append(f'[[ファイル:{media_name}|240px]]')
                     if self._download_media(
                     if self._download_media(
1,394行目: 1,407行目:


                 # videoタグがない場合は取得できない
                 # videoタグがない場合は取得できない
                 video: Final[Tag | None] = video_container.select_one('video')
                 video: Tag | None = video_container.select_one('video')
                 if video is None:
                 if video is None:
                     logger.error(f'{tweet_url}の動画が取得できませんでしたを 当職無能')
                     logger.error(f'{tweet_url}の動画が取得できませんでしたを 当職無能')
1,401行目: 1,414行目:


                 # videoタグのdata-url属性またはvideoタグ直下のsourceタグからURLが取得できる
                 # videoタグのdata-url属性またはvideoタグ直下のsourceタグからURLが取得できる
                 data_url: Final[str | list[str] | None] = video.get('data-url')
                 data_url: str | list[str] | None = video.get('data-url')
                 source_tag: Final[Tag | None] = video.select_one('source')
                 source_tag: Tag | None = video.select_one('source')
                 src_url: Final[str | list[str] | None] = \
                 src_url: str | list[str] | None = \
                     source_tag.get('src') if source_tag is not None else None
                     source_tag.get('src') if source_tag is not None else None
                 video_url: Final[str | list[str] | None] = data_url or src_url
                 video_url: str | list[str] | None = data_url or src_url
                 assert isinstance(video_url, str)
                 assert isinstance(video_url, str)
                 tweet_id: Final[str] = tweet_url.split('/')[-1]
                 tweet_id: str = tweet_url.split('/')[-1]
                 if data_url is not None:
                 if data_url is not None:
                     # data-url属性からURLを取得した場合
                     # data-url属性からURLを取得した場合
                     video_matched: Final[re.Match[str] | None] = re.search(
                     video_matched: re.Match[str] | None = re.search(
                         r'[^/]+$', video_url)
                         r'[^/]+$', video_url)
                     assert video_matched is not None
                     assert video_matched is not None
                     media_path: Final[str] = unquote(video_matched.group())
                     media_path: str = unquote(video_matched.group())
                     media_url: str = urljoin(self.NITTER_INSTANCE, media_path)
                     media_url: str = urljoin(self.NITTER_INSTANCE, media_path)
                     ts_filename: str | None = (
                     ts_filename: str | None = (
1,422行目: 1,435行目:
                     media_url: str = video_url
                     media_url: str = video_url
                     ts_filename: str | None = None
                     ts_filename: str | None = None
                 mp4_filename: Final[str] = (
                 mp4_filename: str = (
                     f'{os.path.join(self.MEDIA_DIR, tweet_id)}_{i}.mp4'
                     f'{os.path.join(self.MEDIA_DIR, tweet_id)}_{i}.mp4'
                 )
                 )
1,491行目: 1,504行目:
             assert poll_info is not None
             assert poll_info is not None
             for poll_meter in poll_meters:
             for poll_meter in poll_meters:
                 poll_choice_value: Final[Tag | None] = poll_meter.select_one(
                 poll_choice_value: Tag | None = poll_meter.select_one(
                     '.poll-choice-value')
                     '.poll-choice-value')
                 assert poll_choice_value is not None
                 assert poll_choice_value is not None
                 ratio: Final[str] = poll_choice_value.text
                 ratio: str = poll_choice_value.text
                 poll_choice_option: Final[Tag | None] = poll_meter.select_one(
                 poll_choice_option: Tag | None = poll_meter.select_one(
                     '.poll-choice-option')
                     '.poll-choice-option')
                 assert poll_choice_option is not None
                 assert poll_choice_option is not None
1,554行目: 1,567行目:
         urls_in_tweet: Final[ResultSet[Tag]] = tag.find_all('a')
         urls_in_tweet: Final[ResultSet[Tag]] = tag.find_all('a')
         for url in urls_in_tweet:
         for url in urls_in_tweet:
             href: Final[str | list[str] | None] = url.get('href')
             href: str | list[str] | None = url.get('href')
             assert isinstance(href, str)
             assert isinstance(href, str)


1,574行目: 1,587行目:
                         and self._invidious_pattern.search(href)):
                         and self._invidious_pattern.search(href)):
                     # Nitter上のYouTubeへのリンクをInvidiousのものから直す
                     # Nitter上のYouTubeへのリンクをInvidiousのものから直す
                     invidious_href: Final[str | list[str] | None] = (
                     invidious_href: str | list[str] | None = (
                         self._invidious_pattern.sub(
                         self._invidious_pattern.sub(
                             'youtube.com' if (
                             'youtube.com' if (
1,594行目: 1,607行目:
             elif url.text.startswith('@'):
             elif url.text.startswith('@'):
                 url_link: str = urljoin(self.TWITTER_URL, href)
                 url_link: str = urljoin(self.TWITTER_URL, href)
                 url_text: Final[str] = url.text
                 url_text: str = url.text
                 url.replace_with(
                 url.replace_with(
                     self._archive_url(
                     self._archive_url(
1,600行目: 1,613行目:


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


1,693行目: 1,707行目:
         tweets: Final[list[Tag]] = self._get_timeline_items(soup)
         tweets: Final[list[Tag]] = self._get_timeline_items(soup)
         for tweet in tweets:
         for tweet in tweets:
             tweet_a: Final[Tag | None] = tweet.a
             tweet_a: Tag | None = tweet.a
             assert tweet_a is not None
             assert tweet_a is not None
             if tweet_a.text == self.NEWEST:
             if tweet_a.text == self.NEWEST:
1,714行目: 1,728行目:
                     continue
                     continue


             tweet_link: Final[Tag | NavigableString | None] = tweet.find(
             tweet_link: Tag | NavigableString | None = tweet.find(
                 class_='tweet-link')
                 class_='tweet-link')
             assert isinstance(tweet_link, Tag)
             assert isinstance(tweet_link, Tag)
             href: Final[str | list[str] | None] = tweet_link.get('href')
             href: str | list[str] | None = tweet_link.get('href')
             assert isinstance(href, str)
             assert isinstance(href, str)
             tweet_url: Final[str] = urljoin(
             tweet_url: str = urljoin(
                 self.TWITTER_URL,
                 self.TWITTER_URL,
                 self._url_fragment_pattern.sub('', href))
                 self._url_fragment_pattern.sub('', href))


             # 日付の更新処理
             # 日付の更新処理
             date: Final[datetime] = self._tweet_date(tweet_url)
             date: datetime = self._tweet_date(tweet_url)
             self._table_builder.next_day_if_necessary(date)
             self._table_builder.next_day_if_necessary(date)


             tweet_callinshow_template: Final[str] = self._callinshowlink_url(
             tweet_callinshow_template: str = self._callinshowlink_url(
                 tweet_url, accessor)
                 tweet_url, accessor)
             tweet_content: Final[Tag | NavigableString | None] = tweet.find(
             tweet_content: Tag | NavigableString | None = tweet.find(
                 class_='tweet-content media-body')
                 class_='tweet-content media-body')
             assert isinstance(tweet_content, Tag)
             assert isinstance(tweet_content, Tag)
             self._archive_soup(tweet_content, accessor)
             self._archive_soup(tweet_content, accessor)
             media_txt: Final[str] = self._fetch_tweet_media(
             media_txt: str = self._fetch_tweet_media(
                 tweet, tweet_url, accessor)
                 tweet, tweet_url, accessor)
             quote_txt: Final[str] = self._get_tweet_quote(tweet, accessor)
             quote_txt: str = self._get_tweet_quote(tweet, accessor)
             poll_txt: Final[str] = self._get_tweet_poll(tweet)
             poll_txt: str = self._get_tweet_poll(tweet)
             self._table_builder.append(
             self._table_builder.append(
                 tweet_callinshow_template, '<br>\n'.join(
                 tweet_callinshow_template, '<br>\n'.join(
1,778行目: 1,792行目:
         new_url: str = ''
         new_url: str = ''
         for show_more in show_mores:  # show-moreに次ページへのリンクか前ページへのリンクがある
         for show_more in show_mores:  # show-moreに次ページへのリンクか前ページへのリンクがある
             show_more_a: Final[Tag | None] = show_more.a
             show_more_a: Tag | None = show_more.a
             assert show_more_a is not None
             assert show_more_a is not None
             href: Final[str | list[str] | None] = show_more_a.get('href')
             href: str | list[str] | None = show_more_a.get('href')
             assert isinstance(href, str)
             assert isinstance(href, str)
             new_url = urljoin(
             new_url = urljoin(
1,805行目: 1,819行目:


     def _signal_handler(
     def _signal_handler(
            self, signum: int, frame: FrameType | None) -> NoReturn:
        self, signum: int, frame: FrameType | None
    ) -> NoReturn:
         """ユーザがCtrl + Cでプログラムを止めたときのシグナルハンドラ。
         """ユーザがCtrl + Cでプログラムを止めたときのシグナルハンドラ。


1,822行目: 1,837行目:
                 'See also: https://nitter.cz'
                 'See also: https://nitter.cz'
                 '\033[0m')
                 '\033[0m')
     def execute(self, krsw: str | None = None, use_browser: bool = True,
     def execute(
                enable_javascript: bool = True) -> None:
        self,
        krsw: str | None = None,
        use_browser: bool = True,
        enable_javascript: bool = True
    ) -> None:
         """通信が必要な部分のロジック。
         """通信が必要な部分のロジック。


1,864行目: 1,883行目:
                     if not self._go_to_new_page(accessor):
                     if not self._go_to_new_page(accessor):
                         break
                         break
             except BaseException as e:
             except BaseException:
                 logger.critical('予想外のエラーナリ。ここまでの成果をダンプして終了するナリ。')
                 logger.critical('予想外のエラーナリ。ここまでの成果をダンプして終了するナリ。')
                 self._table_builder.dump_file()
                 self._table_builder.dump_file()
                 raise e
                 raise




1,891行目: 1,910行目:
         * ちゃんとテストする。
         * ちゃんとテストする。
     """
     """
    WIKI_URL: Final[str] = 'https://krsw-wiki.in'
    """Final[str]: WikiのURL。"""


     TEMPLATE_URL: Final[str] = 'https://krsw-wiki.org/wiki/テンプレート:降臨ショー恒心ログ'
     TEMPLATE_URL: Final[str] = WIKI_URL + '/wiki/テンプレート:降臨ショー恒心ログ'
     """Final[str]: テンプレート:降臨ショー恒心ログのURL。"""
     """Final[str]: テンプレート:降臨ショー恒心ログのURL。"""


1,901行目: 1,922行目:
     @override
     @override
     def _set_queries(
     def _set_queries(
            self, accessor: AccessorHandler, krsw: str | None) -> bool:
        self, accessor: AccessorHandler, krsw: str | None
    ) -> bool:
         """検索条件を設定する。
         """検索条件を設定する。


1,994行目: 2,016行目:


         urls: Final[list[str]] = list(map(
         urls: Final[list[str]] = list(map(
             lambda x: 'https://krsw-wiki.org' + assert_get(x, 'href'),
             lambda x: self.WIKI_URL + assert_get(x, 'href'),
             template_soup.select('.wikitable > tbody > tr > td a')))
             template_soup.select('.wikitable > tbody > tr > td a')))
         for url in urls:
         for url in urls:
             logger.info(f'{unquote(url)} で収集済みツイートを探索中でふ')
             logger.info(f'{unquote(url)} で収集済みツイートを探索中でふ')
             page: Final[str | None] = (
             page: str | None = accessor.request_with_requests_module(url)
                accessor.request_with_requests_module(url))
             assert page is not None
             assert page is not None
             soup: Final[BeautifulSoup] = BeautifulSoup(page, 'html.parser')
             soup: BeautifulSoup = BeautifulSoup(page, 'html.parser')
             url_as = soup.select('tr > th a')
             url_as = soup.select('tr > th a')
             assert len(url_as) > 0
             assert len(url_as) > 0
2,022行目: 2,043行目:
         for tweet in tweets:
         for tweet in tweets:
             # リダイレクトがあればすべてのリンクを、ないなら目的のURLだけが取得できる
             # リダイレクトがあればすべてのリンクを、ないなら目的のURLだけが取得できる
             urls: Final[list[str]] = list(map(
             urls: list[str] = list(map(
                 lambda a: a.text, tweet.select('a')[1:]))
                 lambda a: a.text, tweet.select('a')[1:]))
             # 別のユーザへのリダイレクトがあるものは除く
             # 別のユーザへのリダイレクトがあるものは除く
2,039行目: 2,060行目:
                     logger.debug(url_matched.string + 'は最古の探索対象よりも古いのでポア')
                     logger.debug(url_matched.string + 'は最古の探索対象よりも古いのでポア')
                     continue
                     continue
                 a_first_child: Final[Tag | None] = tweet.select_one(
                 a_first_child: Tag | None = tweet.select_one('a:first-child')
                    'a:first-child')
                 assert a_first_child is not None
                 assert a_first_child is not None
                 archive_url: Final[str | list[str] | None] = a_first_child.get(
                 archive_url: str | list[str] | None = a_first_child.get('href')
                    'href')
                 assert isinstance(archive_url, str)
                 assert isinstance(archive_url, str)
                 if url_matched[0] not in self._url_list_on_wiki:
                 if url_matched[0] not in self._url_list_on_wiki:
2,059行目: 2,078行目:


     def _fetch_next_page(
     def _fetch_next_page(
            self, soup: Tag, accessor: AccessorHandler) -> str | None:
        self, soup: Tag, accessor: AccessorHandler
    ) -> str | None:
         """archive.todayの検索結果のページをpaginateする。
         """archive.todayの検索結果のページをpaginateする。


2,089行目: 2,109行目:
         while has_next:
         while has_next:
             self._append_tweet_urls(soup)
             self._append_tweet_urls(soup)
             next_page: Final[str | None] = self._fetch_next_page(
             next_page: str | None = self._fetch_next_page(soup, accessor)
                soup, accessor)
             if next_page is not None:
             if next_page is not None:
                 soup = BeautifulSoup(next_page, 'html.parser')
                 soup = BeautifulSoup(next_page, 'html.parser')
2,097行目: 2,116行目:


     def _next_url(
     def _next_url(
            self,
        self,
            accessor: AccessorHandler,
        accessor: AccessorHandler,
            tweet_url_prefix: str,
        tweet_url_prefix: str,
            incremented_num: int,
        incremented_num: int,
            incremented: bool = False) -> None:
        incremented: bool = False
    ) -> None:
         """ツイートのURLを、数字部分をインクリメントしながら探索する。
         """ツイートのURLを、数字部分をインクリメントしながら探索する。


2,179行目: 2,199行目:
                           True)
                           True)


     def _parse_images(self, soup: Tag,
     def _parse_images(
                      accessor: AccessorHandler) -> tuple[str, ...]:
        self, soup: Tag, accessor: AccessorHandler
    ) -> tuple[str, ...]:
         """ツイートの魚拓から画像をダウンロードし、ファイル名のタプルを返す。
         """ツイートの魚拓から画像をダウンロードし、ファイル名のタプルを返す。


2,232行目: 2,253行目:
         if len(internal_a_tags) > 0:
         if len(internal_a_tags) > 0:
             for a_tag in internal_a_tags:
             for a_tag in internal_a_tags:
                 account_name: Final[str] = a_tag.text
                 account_name: str = a_tag.text
                 if account_name.startswith('#'):
                 if account_name.startswith('#'):
                     a_tag.replace_with(a_tag.text)
                     a_tag.replace_with(a_tag.text)
                 else:
                 else:
                     url: Final[str] = urljoin(
                     url: str = urljoin(self.TWITTER_URL, account_name[1:])
                        self.TWITTER_URL, account_name[1:])
                     a_tag.replace_with(TableBuilder.archive_url(
                     a_tag.replace_with(TableBuilder.archive_url(
                         url,
                         url,
2,333行目: 2,353行目:


     def _get_tweet_from_archive(
     def _get_tweet_from_archive(
            self,
        self,
            url_pairs: list[UrlTuple],
        url_pairs: list[UrlTuple],
            accessor: AccessorHandler) -> None:
        accessor: AccessorHandler
    ) -> None:
         """魚拓からツイート本文を取得する。
         """魚拓からツイート本文を取得する。


2,345行目: 2,366行目:


         for url_pair in url_pairs:
         for url_pair in url_pairs:
             page: Final[str | None] = accessor.request(url_pair.archive_url)
             page: str | None = accessor.request(url_pair.archive_url)
             assert page is not None
             assert page is not None
             article: Final[Tag | None] = BeautifulSoup(
             article: Tag | None = BeautifulSoup(
                 page, 'html.parser'
                 page, 'html.parser'
             ).select_one('article[tabindex="-1"]')
             ).select_one('article[tabindex="-1"]')
2,355行目: 2,376行目:


                 logger.debug(url_pair.url + 'を整形しますを')
                 logger.debug(url_pair.url + 'を整形しますを')
                 tweet_date: Final[datetime] = self._tweet_date(url_pair.url)
                 tweet_date: datetime = self._tweet_date(url_pair.url)
                 table_builder.next_day_if_necessary(tweet_date)
                 table_builder.next_day_if_necessary(tweet_date)


2,375行目: 2,396行目:


                     # YouTube等のリンク
                     # YouTube等のリンク
                     card_tag: Final[Tag | None] = article.select_one(
                     card_tag: Tag | None = article.select_one(
                         'div[aria-label="Play"]:not(div[role="link"] '
                         'div[aria-label="Play"]:not(div[role="link"] '
                         'div[aria-label="Play"])')
                         'div[aria-label="Play"])')
2,384行目: 2,405行目:


                     # 画像に埋め込まれた外部サイトへのリンク
                     # 画像に埋め込まれた外部サイトへのリンク
                     article_tag: Final[Tag | None] = article.select_one(
                     article_tag: Tag | None = article.select_one(
                         'a[role="link"][aria-label] img:not(div[role="link"] '
                         'a[role="link"][aria-label] img:not(div[role="link"] '
                         'a[role="link"][aria-label] img)')
                         'a[role="link"][aria-label] img)')
2,393行目: 2,414行目:


                     # 引用の有無のチェック
                     # 引用の有無のチェック
                     retweet_tag: Final[Tag | None] = article.select_one(
                     retweet_tag: Tag | None = article.select_one(
                         'div[role="link"]')
                         'div[role="link"]')
                     if retweet_tag is not None:
                     if retweet_tag is not None:
                         account_name_tag: Final[Tag | None] = (
                         account_name_tag: Tag | None = (
                             retweet_tag.select_one(
                             retweet_tag.select_one(
                                 'div[tabindex="-1"] > div > span:not(:has(> *))'))  # noqa: E501
                                 'div[tabindex="-1"] > div > span:not(:has(> *))'))  # noqa: E501
2,414行目: 2,435行目:


                     # 投票の処理
                     # 投票の処理
                     poll_txt: Final[str] = self._get_tweet_poll(article)
                     poll_txt: str = self._get_tweet_poll(article)
                     text = self._concat_texts(text, poll_txt)
                     text = self._concat_texts(text, poll_txt)


                     # バージョンの処理
                     # バージョンの処理
                     possible_version_text_tags: Final[ResultSet[Tag]] = (
                     possible_version_text_tags: ResultSet[Tag] = (
                         article.select(
                         article.select(
                             'div > span > '
                             'div > span > '
2,431行目: 2,452行目:
                 except Exception as e:
                 except Exception as e:
                     # エラーが起きても止めない
                     # エラーが起きても止めない
                     logger.exception(e)
                     logger.exception('エラーが発生してツイートが取得できませんでしたを', exc_info=True)
                     text = 'エラーが発生してツイートが取得できませんでした\n' + ''.join(
                     text = 'エラーが発生してツイートが取得できませんでした\n' + ''.join(
                         TracebackException.from_exception(e).format())
                         TracebackException.from_exception(e).format())
                 table_builder.append(tweet_callinshow_template, text)
                 table_builder.append(tweet_callinshow_template, text)
             else:
             else:
                 logger.warn(url_pair.url + 'はリツイートなので飛ばすナリ。'
                 logger.warning(url_pair.url + 'はリツイートなので飛ばすナリ。'
                            'URLリスト収集の時点でフィルタできなかった、これはいけない。')
                              'URLリスト収集の時点でフィルタできなかった、これはいけない。')


         table_builder.dump_file()
         table_builder.dump_file()


     @override
     @override
     def execute(self, krsw: str | None = None, use_browser: bool = True,
     def execute(
                enable_javascript: bool = True) -> None:
        self,
        krsw: str | None = None,
        use_browser: bool = True,
        enable_javascript: bool = True
    ) -> None:
         """通信が必要な部分のロジック。
         """通信が必要な部分のロジック。


2,470行目: 2,495行目:
                 sys.exit(1)
                 sys.exit(1)
             # Wikiに既に掲載されているツイートのURLを取得
             # Wikiに既に掲載されているツイートのURLを取得
             self._get_tweet_urls_from_wiki(accessor)
             self._get_tweet_urls_from_wiki(accessor) # Wikiに接続できない時はここをコメントアウト


             # 未掲載のツイートのURLを取得する
             # 未掲載のツイートのURLを取得する
2,477行目: 2,502行目:


             # URL一覧ファイルのダンプ
             # URL一覧ファイルのダンプ
             with codecs.open(self.URL_LIST_FILENAME, 'w', 'utf-8') as f:
             with Path(self.URL_LIST_FILENAME).open('w', encoding='utf-8') as f:
                 for url_pair in self._url_list:
                 for url_pair in self._url_list:
                     f.write(url_pair.url + '\n')
                     f.write(url_pair.url + '\n')
Fet-Fe
66

回編集

「https://krsw-wiki.in/wiki/特別:携帯機器差分/143431」から取得

案内メニュー