唐澤貴洋Wiki

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

利用者ページ 議論

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

2023年11月26日 (日) 16:59時点における版

2,180 バイト追加 、 2023年11月26日 (日)
→‎コード: v4.1.8 Nitterd
>Fet-Fe
(→‎コード: v4.1.7 ユーザがよく変更する定数をUserPropertiesに移動。ページの読み込みが完了するまで待つよう幾つかの箇所を修正)
>Fet-Fe
(→‎コード: v4.1.8 Nitterd)
11行目: 11行目:
"""Twitter自動収集スクリプト
"""Twitter自動収集スクリプト


ver4.1.7 2023/11/18恒心
ver4.1.8 2023/11/26恒心


当コードは恒心停止してしまった https://rentry.co/7298g の降臨ショーツイート自動収集スクリプトの復刻改善版です
当コードは恒心停止してしまった https://rentry.co/7298g の降臨ショーツイート自動収集スクリプトの復刻改善版です
77行目: 77行目:
from typing import (Final, NamedTuple, NoReturn, Self, assert_never, final,
from typing import (Final, NamedTuple, NoReturn, Self, assert_never, final,
                     override)
                     override)
from urllib.parse import quote, unquote, urljoin
from urllib.parse import (ParseResult, parse_qs, quote, unquote, urljoin,
                          urlparse)
from zoneinfo import ZoneInfo
from zoneinfo import ZoneInfo


102行目: 103行目:
     実行時に変更する可能性のある定数はここで定義する。
     実行時に変更する可能性のある定数はここで定義する。
     """
     """
     media_dir: Final[str] = 'tweet_media'
     media_dir: Final[str] = 'tweet_media'
     """Final[str]: 画像などのツイートメディアをダウンロードするディレクトリ。
     """Final[str]: 画像などのツイートメディアをダウンロードするディレクトリ。
    """
    filename: Final[str] = 'tweet.txt'
    """Final[str]: ツイートを保存するファイルの名前。
     """
     """


117行目: 121行目:
         report_interval: Final[int] = 5
         report_interval: Final[int] = 5
         """Final[int]: 記録件数を報告するインターバル。
         """Final[int]: 記録件数を報告するインターバル。
        """
        filename: Final[str] = 'tweet.txt'
        """Final[str]: ツイートを保存するファイルの名前。
         """
         """


160行目: 160行目:
         """
         """


         filename: Final[str] = 'url_list.txt'
         url_list_filename: Final[str] = 'url_list.txt'
         """Final[str]: URLのリストをダンプするファイル名。
         """Final[str]: URLのリストをダンプするファイル名。
         """
         """
213行目: 213行目:
         Returns:
         Returns:
             str: レスポンスのHTML。
             str: レスポンスのHTML。
        """
        ...
    @abstractmethod
    def set_cookies(self, cookies: dict[str, str]) -> None:
        """Cookieをセットする。
        Args:
            cookies (dict[str, str]): Cookieのキーバリューペア。
         """
         """
         ...
         ...
259行目: 268行目:
         """
         """
         # Ｔorに必要なプロキシをセット
         # Ｔorに必要なプロキシをセット
        self._cookies: dict[str, str] = {}
         self._proxies: dict[str, str] | None = self._choose_tor_proxies()
         self._proxies: dict[str, str] | None = self._choose_tor_proxies()


268行目: 278行目:


         Raises:
         Raises:
             requests.exceptions.HTTPError: ステータスコードが200でない場合のエラー。
             requests.HTTPError: ステータスコードが200でない場合のエラー。


         Returns:
         Returns:
279行目: 289行目:
             headers=self.HEADERS,
             headers=self.HEADERS,
             allow_redirects=False,
             allow_redirects=False,
             proxies=self._proxies)
             proxies=self._proxies,
            cookies=self._cookies)
         res.raise_for_status()  # HTTPステータスコードが200番台以外でエラー発生
         res.raise_for_status()  # HTTPステータスコードが200番台以外でエラー発生
         return res
         return res
287行目: 298行目:
         try:
         try:
             return self._execute(url).text
             return self._execute(url).text
         except requests.exceptions.HTTPError as e:
         except (requests.HTTPError, requests.ConnectionError) as e:
             raise AccessError from e
             raise AccessError from e
    @override
    def set_cookies(self, cookies: dict[str, str]) -> None:
        self._cookies.update(cookies)


     def get_image(self, url: str) -> bytes | None:
     def get_image(self, url: str) -> bytes | None:
304行目: 319行目:
         try:
         try:
             res: Final[requests.models.Response] = self._execute(url)
             res: Final[requests.models.Response] = self._execute(url)
         except requests.exceptions.HTTPError as e:
         except (requests.HTTPError, requests.ConnectionError) as e:
             raise AccessError from e
             raise AccessError from e


403行目: 418行目:
             self.TOR_BROWSER_PATHS[platform.system()]
             self.TOR_BROWSER_PATHS[platform.system()]
         )
         )
        self._options.add_argument('--user-data-dir=selenium')  # type: ignore


         if enable_javascript:
         if enable_javascript:
456行目: 472行目:
             ReCaptchaRequiredError: JavaScriptがオフの状態でreCAPTCHAが要求された場合のエラー。
             ReCaptchaRequiredError: JavaScriptがオフの状態でreCAPTCHAが要求された場合のエラー。
         """
         """
         iframe_selector: Final[str] = \
         iframe_selector: Final[str] = (
             'iframe[title="recaptcha challenge expires in two minutes"]'
             'iframe[title="recaptcha challenge expires in two minutes"]')


         if len(self._driver.find_elements(By.ID, 'g-recaptcha')) > 0:
         if len(self._driver.find_elements(By.ID, 'g-recaptcha')) > 0:
510行目: 526行目:
             raise AccessError from e
             raise AccessError from e
         return self._driver.page_source
         return self._driver.page_source
    @override
    def set_cookies(self, cookies: dict[str, str]) -> None:
        for name, value in cookies.items():
            self._driver.add_cookie(  # type: ignore
                {'name': name, 'value': value})
        self._driver.refresh()




540行目: 563行目:
         )
         )
         self._requests_accessor: Final[RequestsAccessor] = RequestsAccessor()
         self._requests_accessor: Final[RequestsAccessor] = RequestsAccessor()
        self._last_url = ''  # 最後にアクセスしたURL


     def __enter__(self) -> Self:
     def __enter__(self) -> Self:
602行目: 626行目:
             T | None: レスポンス。接続失敗が何度も起きると `None` を返す。
             T | None: レスポンス。接続失敗が何度も起きると `None` を返す。
         """
         """
        assert url, 'URLが空っぽでふ'
         logger.debug('Requesting ' + unquote(url))
         logger.debug('Requesting ' + unquote(url))
         assert url is not None
         self._last_url = url


         for i in range(self.LIMIT_N_REQUESTS):
         for i in range(self.LIMIT_N_REQUESTS):
670行目: 695行目:
             url,
             url,
             self._requests_accessor.get_image)
             self._requests_accessor.get_image)
    def set_cookies(self, cookies: dict[str, str]) -> None:
        """Cookieをセットする。
        Args:
            cookies (dict[str, str]): Cookieのキーバリューペア。
        """
        self._requests_accessor.set_cookies(cookies)
        if self._selenium_accessor is not None:
            self._selenium_accessor.set_cookies(cookies)
    @property
    def last_url(self) -> str:
        """最後にアクセスしたURLを返す。
        Returns:
            str: 最後にアクセスしたURL。
        """
        return self._last_url


     @property
     @property
685行目: 729行目:
     """
     """


     FILENAME: Final[str] = UserProperties.NitterCrawler.filename
     FILENAME: Final[str] = UserProperties.filename
     """Final[str]: ツイートを保存するファイルの名前。
     """Final[str]: ツイートを保存するファイルの名前。
     """
     """


     def __init__(self, date: datetime) -> None:
     def __init__(self, date: datetime | None = None) -> None:
         """コンストラクタ。
         """コンストラクタ。


         Args:
         Args:
             date (datetime): 記録するツイートの最新日付。
             date (datetime | None, optional): 記録するツイートの最新日付。デフォルトは今日の日付。
         """
         """
         self._tables: Final[list[str]] = ['']
         self._tables: Final[list[str]] = ['']
         self._count: int = 0  # 記録数
         self._count: int = 0  # 記録数
         self._date: datetime = date
         self._date: datetime = date or datetime.today()


     @property
     @property
715行目: 759行目:
             text (str): ツイートの本文。
             text (str): ツイートの本文。
         """
         """
         self._tables[-1] = '!' + callinshow_template + '\n|-\n|\n' \
         self._tables[-1] = (
             + text \
            '!' + callinshow_template + '\n|-\n|\n'
             + '\n|-\n' \
             + text
             + self._tables[-1]
             + '\n|-\n'
             + self._tables[-1])
         self._count += 1
         self._count += 1


742行目: 787行目:


     def _next_day(self, date: datetime | None = None) -> None:
     def _next_day(self, date: datetime | None = None) -> None:
         """Wikiテーブルに日付の見出しを付与する。
         """Wikiテーブルに日付の見出しを付与し、日付を更新する。
 
        記録済みのツイートが1つもない場合は日付の更新だけをする。


         Args:
         Args:
805行目: 852行目:
                         i += len('{{Archive|')
                         i += len('{{Archive|')
                     elif text[i:i + len('https://')] == 'https://':
                     elif text[i:i + len('https://')] == 'https://':
                         text = text[:i] + \
                         text = (text[:i]
                            '<nowiki>https://</nowiki>' + \
                                + '<nowiki>https://</nowiki>'
                            text[i + len('https://'):]
                                + text[i + len('https://'):])
                         i += len('<nowiki>https://</nowiki>')
                         i += len('<nowiki>https://</nowiki>')
                     elif text[i:i + len('http://')] == 'http://':
                     elif text[i:i + len('http://')] == 'http://':
                         text = text[:i] + \
                         text = (text[:i]
                            '<nowiki>http://</nowiki>' + \
                                + '<nowiki>http://</nowiki>'
                            text[i + len('http://'):]
                                + text[i + len('http://'):])
                         i += len('<nowiki>http://</nowiki>')
                         i += len('<nowiki>http://</nowiki>')
                 i += 1
                 i += 1
1,037行目: 1,084行目:
                     + self.TWEETS_OR_REPLIES))
                     + self.TWEETS_OR_REPLIES))
         if page_optional is None:
         if page_optional is None:
             self._on_fail()
             self._on_fail(accessor)
             return False
             return False
         self._page: str = page_optional
         self._nitter_page: str = page_optional
 
        # Nitter用のCookieをセット
        accessor.set_cookies({
            'infiniteScroll': '',
            'proxyVideos': 'on',
            'replaceReddit': '',
            'replaceYouTube': '',
        })


         # 終わりにするツイート取得
         # 終わりにするツイート取得
1,055行目: 1,110行目:
         # 日付取得
         # 日付取得
         timeline_item: Final[Tag | NavigableString | None] = BeautifulSoup(
         timeline_item: Final[Tag | NavigableString | None] = BeautifulSoup(
             self._page, 'html.parser'
             self._nitter_page, 'html.parser'
         ).find(class_='timeline-item')
         ).find(class_='timeline-item')
         assert isinstance(timeline_item, Tag)
         assert isinstance(timeline_item, Tag)
1,184行目: 1,239行目:
                     urljoin(self.NITTER_INSTANCE, account_str))
                     urljoin(self.NITTER_INSTANCE, account_str))
                 if res is None:  # リクエスト失敗判定
                 if res is None:  # リクエスト失敗判定
                     self._on_fail()
                     self._on_fail(accessor)
                     return None
                     return None
                 soup: Final[BeautifulSoup] = BeautifulSoup(res, 'html.parser')
                 soup: Final[BeautifulSoup] = BeautifulSoup(res, 'html.parser')
1,208行目: 1,263行目:
         print('クエリのピースが埋まっていく。')
         print('クエリのピースが埋まっていく。')


     def _on_fail(self) -> None:
     def _on_fail(self, accessor: AccessorHandler) -> None:
         """接続失敗時処理。
         """接続失敗時処理。


         取得に成功した分だけファイルにダンプする。
         取得に成功した分だけファイルにダンプする。
        ログにトレースバックも表示する。
        Args:
            accessor (AccessorHandler): アクセスハンドラ。
         """
         """
         logger.critical('接続失敗時処理をしておりまふ')
         logger.critical('接続失敗時処理をしておりまふ', stack_info=True)
        logger.critical('最後にアクセスしたURL: ' + accessor.last_url)
         print('接続失敗しすぎで強制終了ナリ')
         print('接続失敗しすぎで強制終了ナリ')
         if self._table_builder.count > 0:  # 取得成功したデータがあれば発行
         if self._table_builder.count > 0:  # 取得成功したデータがあれば発行
1,664行目: 1,724行目:
         """
         """
         soup: Final[BeautifulSoup] = BeautifulSoup(
         soup: Final[BeautifulSoup] = BeautifulSoup(
             self._page, 'html.parser')
             self._nitter_page, 'html.parser')
         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:
1,708行目: 1,768行目:
             self._archive_soup(tweet_content, accessor)
             self._archive_soup(tweet_content, accessor)
             media_txt: Final[str] = self._fetch_tweet_media(
             media_txt: Final[str] = self._fetch_tweet_media(
                 tweet,
                 tweet, tweet_url, accessor)
                tweet_url,
                accessor)
             quote_txt: Final[str] = self._get_tweet_quote(tweet, accessor)
             quote_txt: Final[str] = self._get_tweet_quote(tweet, accessor)
             poll_txt: Final[str] = self._get_tweet_poll(tweet)
             poll_txt: Final[str] = self._get_tweet_poll(tweet)
1,747行目: 1,805行目:
             bool: 次のページを取得できれば `True`。
             bool: 次のページを取得できれば `True`。
         """
         """
         soup: Final[BeautifulSoup] = BeautifulSoup(self._page, 'html.parser')
         soup: Final[BeautifulSoup] = BeautifulSoup(self._nitter_page,
                                                  'html.parser')
         show_mores: Final[ResultSet[Tag]] = soup.find_all(class_='show-more')
         show_mores: Final[ResultSet[Tag]] = soup.find_all(class_='show-more')
        assert len(show_mores) > 0
         new_url: str = ''
         new_url: str = ''
         for show_more in show_mores:  # show-moreに次ページへのリンクか前ページへのリンクがある
         for show_more in show_mores:  # show-moreに次ページへのリンクか前ページへのリンクがある
             if show_more.text != self.NEWEST:  # 前ページへのリンクではないか判定
             show_more_a: Final[Tag | None] = show_more.a
                show_more_a: Final[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: Final[str | list[str] | None] = show_more_a.get('href')
            assert isinstance(href, str)
                assert isinstance(href, str)
            new_url = urljoin(
                new_url = urljoin(
                self.NITTER_INSTANCE,
                    self.NITTER_INSTANCE,
                self._name
                    self._name
                + '/'
                    + '/'
                + self.TWEETS_OR_REPLIES
                    + self.TWEETS_OR_REPLIES
                + href)  # 直下のaタグのhrefの中身取ってURL頭部分と合体
                    + href)  # 直下のaタグのhrefの中身取ってURL頭部分と合体
 
         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._on_fail()
             self._on_fail(accessor)
             return False
             return False
         new_page_soup: Final[BeautifulSoup] = BeautifulSoup(res, 'html.parser')
         new_page_soup: Final[BeautifulSoup] = BeautifulSoup(res, 'html.parser')
1,770行目: 1,831行目:
             # ツイートの終端ではtimeline-endだけのページになるので判定
             # ツイートの終端ではtimeline-endだけのページになるので判定
             logger.info(new_url + 'に移動しますを')
             logger.info(new_url + 'に移動しますを')
             self._page = res  # まだ残りツイートがあるのでページを返して再度ツイート本文収集
             self._nitter_page = res  # まだ残りツイートがあるのでページを返して再度ツイート本文収集
             return True
             return True
         else:
         else:
匿名利用者
「https://krsw-wiki.in/wiki/特別:携帯機器差分/94676」から取得