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

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

2023年4月29日 (土) 16:23時点における版

105 バイト追加、 2023年4月29日 (土)

→‎コード: Twitterの相次ぐ仕様変更のせいでnitterのsearchが使えなくなっていたので、tweets or repliesから取得するようにとりとり変更。またURLの仕様も変わり、もはや前のスクリプトでは取得ができなくなったので、メジャーバージョンアップ。

匿名利用者

>Fet-Fe

@@ 6行目: / 6行目: @@
 '''
-ver2.2.8 2023/3/6恒心
+ver3.0.0 2023/4/29恒心
 当コードは恒心停止してしまったhttps://rentry.co/7298gの降臨ショーツイート自動収集スクリプトの復刻改善版です
@@ 108行目: / 108行目: @@
    ##nitterのURLのドメインとユーザーネーム部分の接続部品
-   SEARCH_QUERY: Final[str] = 'search?q=from%3A'
+   TWEETS_OR_REPLIES: Final[str] = 'with_replies'
    ##HTTPリクエスト時のユーザーエージェントとヘッダ
@@ 128行目: / 128行目: @@
    ##万が一仕様変更で変わったとき用
    NO_ARCHIVE: Final[str] = 'No results'
-  ##nitterで次ページ遷移時のURLから抜け落ちてる部分
-  SEARCH: Final[str] = 'search'
    ##nitterの前ページ読み込み部分の名前
@@ 156行目: / 153行目: @@
      if krsw:
        print('名前は自動的に' + self.CALLINSHOW + 'にナリます')
-       name: Final[str] = self.CALLINSHOW
+       self.name: Final[str] = self.CALLINSHOW
      else:
-       name: Final[str] = self._get_name()
+       self.name: Final[str] = self._get_name()
      ##検索クエリとページ取得
+    self.query_strs: list[str] = []
      if krsw:
        print('クエリは自動的になしにナリます')
-      self._page: Response | None = self._request(urljoin(self.NITTER_INSTANCE, self.SEARCH_QUERY + name))
-      if self._page is None:
-        self._fail()
      else:
-       self._page: Response = self._get_query(name)
+       self._get_query()
+    self._page: Response | None = self._request(urljoin(self.NITTER_INSTANCE, self.name + '/' + self.TWEETS_OR_REPLIES))
+    if self._page is None:
+      self._fail()
      ##終わりにするツイート取得
@@ 323行目: / 321行目: @@
    ##検索クエリを取得
-   def _get_query(self, name: Final[str]) -> Response | NoReturn:
+   def _get_query(self) -> None:
-     while True:
+     print("検索クエリを入れるナリ。複数ある時は一つの塊ごとに入力するナリ。空欄を入れると検索開始ナリ。")
-      print("検索クエリを入れるナリ。複数ある時は一つの塊ごとに入力するナリ。空欄を入れると検索開始ナリ。")
+    print("例：「無能」→改行→「脱糞」→改行→「弟殺し」→改行→空欄で改行")
-      print("例：「無能」→改行→「脱糞」→改行→「弟殺し」→改行→空欄で改行")
+    query_input: str = input() ##ユーザー入力受付
-      query_str: list[str] = [name] ##検索クエリ文字列をquery_strに収納。ユーザー名を先に含めておく
+    ##空欄が押されるまでユーザー入力受付
-      query_input: str = input() ##ユーザー入力受付
+    while query_input != '':
-      ##空欄が押されるまでユーザー入力受付
+      self.query_strs.append(query_input)
-      while query_input != '':
+      query_input = input()
-        query_str.append(quote(query_input))
+    print("クエリのピースが埋まっていく。")
-        query_input = input()
-      print("検索クエリでリクエストしているナリ……")
-      res: Response | None = self._request(urljoin(self.NITTER_INSTANCE, self.SEARCH_QUERY + '+'.join(query_str))) ##リクエストして結果取得
-      if res is None : ##リクエスト失敗判定
-        self._fail()
-      soup: BeautifulSoup = BeautifulSoup(res.text, 'html.parser') ##beautifulsoupでレスポンス解析
-      ##1件もないときはHTMLにtimeline-noneがあるから判定
-      if soup.find(class_='timeline-none') is None:
-        print("クエリのピースが埋まっていく。")
-        return res ##成功したリクエストのレスポンスを返す
-      else:
-        print("適切なクエリを入力することを切に望む。")
    ##接続失敗時処理
@@ 437行目: / 423行目: @@
            # 動画のダウンロード
            if self._proxy_is_needed:
-               returncode: int = subprocess.run(f"yes N 2>/dev/null | ffmpeg -http_proxy {self.PROXIES['http']} -i \"{urljoin(self.NITTER_INSTANCE, media_url)}\" -c copy {os.path.join(self.MEDIA_DIR, tweet_id)}_{i}.ts &> /dev/null", shell=True).returncode
+               returncode: int = subprocess.run(["ffmpeg", "-y", "-http_proxy", "self.PROXIES['http']", "-i", urljoin(self.NITTER_INSTANCE, media_url), "-c", "copy", f"{os.path.join(self.MEDIA_DIR, tweet_id)}_{i}.ts"], stdout=subprocess.DEVNULL).returncode
            else:
-               returncode: int = subprocess.run(f"yes N 2>/dev/null | ffmpeg -i \"{urljoin(self.NITTER_INSTANCE, media_url)}\" -c copy {os.path.join(self.MEDIA_DIR, tweet_id)}_{i}.ts &> /dev/null", shell=True).returncode
+               returncode: int = subprocess.run(["ffmpeg", "-y", "-i", urljoin(self.NITTER_INSTANCE, media_url), "-c", "copy", f"{os.path.join(self.MEDIA_DIR, tweet_id)}_{i}.ts"], stdout=subprocess.DEVNULL).returncode
            # 取得成功したらtsをmp4に変換
            if returncode == 0:
-             ts2mp4_returncode: int = subprocess.run(f"yes N 2>/dev/null | ffmpeg -i {os.path.join(self.MEDIA_DIR, tweet_id)}_{i}.ts -acodec copy -vcodec copy {os.path.join(self.MEDIA_DIR, tweet_id)}_{i}.mp4 &> /dev/null", shell=True).returncode
+             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:
                print(f"{os.path.join(self.MEDIA_DIR, tweet_id)}_{i}.mp4をアップロードしなければない。")
@@ 461行目: / 447行目: @@
        link = re.sub('#.*$', '', link)
        link = urljoin(self.TWITTER_URL, link)
-       quote_txt = self._archive_url(link, link)
+       quote_txt = self._archive_url(link)
      return quote_txt
@@ 503行目: / 489行目: @@
        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 query_strs:
+          if query_str not in tweet.text:
+            not_match = True
+            break
+        if not_match:
+          continue
        tweet_url: str = urljoin(self.TWITTER_URL, re.sub('#[^#]*$', '', tweet.find(class_='tweet-link').get('href'))) ##ツイートのURL作成
        date: datetime = self._tweet_date(tweet)
@@ 513行目: / 510行目: @@
        quote_txt: str = self._get_tweet_quote(tweet) ##引用リツイートの場合、元ツイートを追加
        poll_txt: str = self._get_tweet_poll(tweet) ##投票の取得
-       self._check_additional_info(tweet, tweet_url) ##編集や長いツイートの省略をチェック
+       self._check_additional_info(tweet_content, tweet_url) ##編集や長いツイートの省略をチェック
        self._txt_data[0] = '!' + archived_tweet_url + '\n|-\n|\n' \
          + '<br>\n'.join(filter(None, [
@@ 535行目: / 532行目: @@
    #MediaWiki文法と衝突する文字を無効化する
    def _escape_wiki_reserved_words(self, text: str) -> str:
+    def escape_nolink_urls(text: str) -> str:
+      is_in_archive_template: bool = False
+      i: int = 0
+      while i < len(text):
+        if is_in_archive_template:
+          if text[i:i+2] == '}}':
+            is_in_archive_template = False
+            i += 2
+        else:
+          if text[i:i+10] == '{{Archive|' or text[i:i+10] == '{{archive|':
+            is_in_archive_template = True
+            i += 10
+          elif text[i:i+8] == 'https://':
+            text = text[:i] + '<nowiki>https://</nowiki>' + text[i+8:]
+            i += 25
+          elif text[i:i+7] == 'http://':
+            text = text[:i] + '<nowiki>http://</nowiki>' + text[i+7:]
+            i += 24
+        i += 1
+      return text
      text = text.replace('\n', '<br>\n')
      text = re.sub(r'^ ', '&nbsp;', text, flags=re.MULTILINE)
      text = re.sub(r'^([\*#:;])', r'<nowiki>\1</nowiki>', text, flags=re.MULTILINE)
      text = re.sub(r'^----', '<nowiki>----</nowiki>', text, flags=re.MULTILINE)
+    text = escape_nolink_urls(text)
      return text
-   #tagをテンプレートArchiveの文字列に変化させる
+   #aタグをテンプレートArchiveの文字列に変化させる
    def _archive_soup(self, tag: bs4.element.Tag) -> None:
      urls_in_tweet: Final[bs4.element.ResultSet] = tag.find_all('a')
@@ 550行目: / 569行目: @@
            url_link: str = url.get('href').replace('https' + self.NITTER_INSTANCE[4:], self.TWITTER_URL)
            url_link = re.sub('\?.*$', '', url_link)
-           url.replace_with(self._archive_url(url_link, url_link)) ##テンプレートArchiveに変化
+           url.replace_with(self._archive_url(url_link)) ##テンプレートArchiveに変化
          elif url.get('href').startswith('https://nitter.kavin.rocks/'):
            #Nitter上のTwitterへのリンクを直す
            url_link: str = url.get('href').replace('https://nitter.kavin.rocks/', self.TWITTER_URL)
            url_link = re.sub('\?.*$', '', url_link)
-           url.replace_with(self._archive_url(url_link, url_link)) ##テンプレートArchiveに変化
+           url.replace_with(self._archive_url(url_link)) ##テンプレートArchiveに変化
          elif self._invidious_pattern.search(url.get('href')):
            #Nitter上のYouTubeへのリンクをInvidiousのものから直す
            url_link: str = url.get('href')
-           if re.match('https://[^/]+/[^/]+/', url_link):
+           if re.match('https://[^/]+/[^/]+/', url_link) or re.search('/@[^/]*$', url_link):
              url_link = self._invidious_pattern.sub('youtube.com', url_link)
            else:
              url_link = self._invidious_pattern.sub('youtu.be', url_link)
-           url.replace_with(self._archive_url(url_link, url_link)) ##テンプレートArchiveに変化
+           url.replace_with(self._archive_url(url_link)) ##テンプレートArchiveに変化
          elif url.get('href').startswith('https://bibliogram.art/'):
            # Nitter上のInstagramへのリンクをBibliogramのものから直す
            # Bibliogramは中止されたようなのでそのうちリンクが変わるかも
            url_link: str = url.get('href').replace('https://bibliogram.art/', 'https://www.instagram.com/')
-           url.replace_with(self._archive_url(url_link, url_link)) ##テンプレートArchiveに変化
+           url.replace_with(self._archive_url(url_link)) ##テンプレートArchiveに変化
          else:
-           url.replace_with(self._archive_url(url.get('href'), url.get('href'))) ##テンプレートArchiveに変化
+           url.replace_with(self._archive_url(url.get('href'))) ##テンプレートArchiveに変化
-   #URLをテンプレートArchiveに変化させる
+   #URLをテンプレートArchiveに変化させる。#がURLに含まれていたら別途処理
-   def _archive_url(self, url: Final[str], text: Final[str]) -> str:
+   def _archive_url(self, url: Final[str]) -> str:
-     return '{{Archive|1=' + unquote(url) + '|2=' + self._archive(url) + '}}' ##テンプレートArchiveの文字列返す
+     if '#' in url:
+      main_url, fragment = url.split('#', maxsplit=1)
+      return '{{Archive|1=' + unquote(url) + '|2=' + self._archive(main_url) + '#' + fragment + '}}' ##テンプレートArchiveの文字列返す
+    else:
+      return '{{Archive|1=' + unquote(url) + '|2=' + self._archive(url) + '}}' ##テンプレートArchiveの文字列返す
    #URLをテンプレートCallinShowlinkに変化させる
@@ 582行目: / 605行目: @@
    ##URLから魚拓返す
    def _archive(self, url: Final[str]) -> str:
-    def encode_query_value_equals(url: str) -> str:
-      query_found: bool = False
-      in_query_value: bool = False
-      i: int = 0
-      for c in url:
-        if c == '?':
-          query_found = True
-        elif query_found and (not in_query_value) and c == '=':
-          in_query_value = True
-        elif query_found and in_query_value and c == '=':
-          url = url[:i] + '%3D' + url[i+1:]
-          i += 2
-        elif query_found and in_query_value and c == '&':
-          in_query_value = False
-        i += 1
-      return url
      archive_url: str = urljoin(self.ARCHIVE_TODAY_STANDARD, quote(unquote(url), safe='&=+?%')) ##wikiに載せるとき用URLで失敗するとこのままhttps://archive.ph/https%3A%2F%2Fxxxxxxxxの形で返される
      res: Final[Response | None] = self._request(urljoin(self.ARCHIVE_TODAY, quote(unquote(url), safe='&=+?%'))) ##アクセス用URL使って結果を取得
@@ 616行目: / 622行目: @@
      soup: Final[BeautifulSoup] = BeautifulSoup(self._page.text, 'html.parser') ##beautifulsoupでレスポンス解析
      show_mores: Final[bs4.element.ResultSet] = soup.find_all(class_="show-more")
+    new_url: str = '' # ここで定義しないと動かなくなった、FIXME?
      for show_more in show_mores: ##show-moreに次ページへのリンクか前ページへのリンクがある
        if show_more.text != self.NEWEST:  ##前ページへのリンクではないか判定
-         new_url: str = urljoin(self.NITTER_INSTANCE, self.SEARCH + show_more.a.get('href')) ##直下のaタグのhrefの中身取ってURL頭部分と合体
+         new_url = urljoin(self.NITTER_INSTANCE, self.CALLINSHOW + '/' + self.TWEETS_OR_REPLIES + show_more.a.get('href')) ##直下のaタグのhrefの中身取ってURL頭部分と合体
      res: Final[Response | None] = self._request(new_url) ##接続してHTML取ってくる
      if res is None: