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

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

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

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

2022年8月9日 (火) 01:31時点における版

5,214 バイト追加 、 2022年8月9日 (火)
→‎コード: ツイートの画像を自動取得するため、勝手ながら改造させていただきました
>夜泣き
編集の要約なし
>Fet-Fe
(→‎コード: ツイートの画像を自動取得するため、勝手ながら改造させていただきました)
6行目: 6行目:


'''
'''
ver2.1.0 2022/8/8恒心
当コードは恒心停止してしまったhttps://rentry.co/7298gの降臨ショーツイート自動収集スクリプトの復刻改善版です
当コードは恒心停止してしまったhttps://rentry.co/7298gの降臨ショーツイート自動収集スクリプトの復刻改善版です
前開発者との出会いに感謝
前開発者との出会いに感謝
22行目: 24行目:


・環境は玉葱前提です。
・環境は玉葱前提です。
・Whonix-Workstationで動作確認済
・Whonix-Workstation, MacOSで動作確認済
・bs4はインストールしないと標準で入ってません
  ・MacOSの場合はbrewでtorコマンドを導入し、実行
・PySocks, bs4はインストールしないと標準で入ってません
・requestsも環境によっては入っていないかもしれない
・requestsも環境によっては入っていないかもしれない
・$ pip install bs4 requests
・$ pip install bs4 requests PySocks
・pipも入っていなければ$ sudo apt install pip
・pipも入っていなければ$ sudo apt install pip
・バグ報告はhttps://krsw-wiki.org/wiki/?curid=15799
・バグ報告はhttps://krsw-wiki.org/wiki/?curid=15799#利用者:夜泣き/スクリプトについて


ーーーーーーーーーーーーーーーーーーーーーーーーーーーーーーー
ーーーーーーーーーーーーーーーーーーーーーーーーーーーーーーー
34行目: 37行目:
#インポート類
#インポート類
import sys
import sys
import os
import codecs
import codecs
import re
import re
import json
from time import sleep
from time import sleep
from typing import Final, Optional, Union, NoReturn, ClassVar
from urllib.parse import quote, unquote
import warnings
import requests
import requests
from bs4 import BeautifulSoup as bs4
import bs4
from typing import Final
from bs4 import BeautifulSoup
from typing import Optional
from urllib.parse import quote
import warnings


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


##nitterのインスタンス
class TwitterArchiver:
##生きているのはhttps://github.com/zedeus/nitter/wiki/Instancesで確認
  #定数・設定類
Nitterinstance = 'http://nitterqdyumlovt7tjqpdjrluitgmtpa53qq3idlpgoe4kxo7gs3xvad.onion/'
 
  ##nitterのインスタンス
  ##生きているのはhttps://github.com/zedeus/nitter/wiki/Instancesで確認
  ##末尾にスラッシュ必須
  NITTER_INSTANCE: Final[str] = 'http://ibsboeui2im5o7dxnik3s5yghufumgy5abevtij5nbizequfpu4qi4ad.onion/'
 
  ##archive.todayの魚拓
  ##実際にアクセスして魚拓があるか調べるのにはonionを使用
  ##末尾にスラッシュ必須
  ARCHIVE_TODAY: Final[str] = 'http://archiveiya74codqgiixo33q62qlrqtkgmcitqx5u2oeqnmn5bpcbiyd.onion/'
 
  ##archive.todayの魚拓
  ##記事の文章に使うのはクリアネット
  ##末尾にスラッシュ必須
  ARCHIVE_TODAY_STANDARD: Final[str] = 'https://archive.ph/'
 
  ##twitterのURL
  ##末尾にスラッシュ禁止
  TWITTER_URL: Final[str] = 'https://twitter.com'
 
  ##降臨ショーのユーザーネーム
  CALLINSHOW: Final[str] = 'CallinShow'
 
  ##HTTPリクエストのタイムアウト秒数
  REQUEST_TIMEOUT: Final[int] = 30


##archive.todayの魚拓
  ##取得するツイート数の上限
##実際にアクセスして魚拓があるか調べるのにはonionを使用
  LIMIT_N_TWEETS: Final[int] = 300
Archivetoday  = 'http://archiveiya74codqgiixo33q62qlrqtkgmcitqx5u2oeqnmn5bpcbiyd.onion/'


##archive.todayの魚拓
  ##記録件数を報告するインターバル
##記事の文章に使うのはクリアネット
  REPORT_INTERVAL: Final[int] = 5
Archivetodaystandard = 'https://archive.ph/'


##twitterのURL
  ##HTTPリクエスト失敗時の再試行回数
Twitterurl = 'https://twitter.com/'
  LIMIT_N_REQUESTS: Final[int] = 5


##降臨ショーのユーザーネーム
  ##HTTPリクエスト失敗時にさらに追加する待機時間
Callinshow: Final[str] = 'CallinShow'
  WAIT_TIME_FOR_ERROR: Final[int] = 4


##HTTPリクエストのタイムアウト秒数
  ##HTTPリクエスト成功失敗関わらず待機時間
Request_timeout: Final[int] = 30
  ##1秒待つだけで行儀がいいクローラーだそうなので既定では1秒
  ##しかし日本のポリホーモは1秒待っていても捕まえてくるので注意
  ##https://ja.wikipedia.org/wiki/?curid=2187212
  WAIT_TIME: Final[int] = 1


##取得するツイート数の上限
  ##nitterのURLのドメインとユーザーネーム部分の接続部品
Limit_tweet: Final[int] = 100
  SEARCH_QUERY: Final[str] = 'search?q=from%3A'


##HTTPリクエスト失敗時の再試行回数
  ##HTTPリクエスト時のユーザーエージェントとヘッダ
Limit_request: Final[int] = 5
  HEADERS: Final[dict[str, str]] = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; rv:91.0) Gecko/20100101 Firefox/91.0'
  }


##HTTPリクエスト失敗時にさらに追加する待機時間
  ##HTTP
Waittime_error: Final[int] = 4
  PROXIES: Final[dict[str, str]] = {
    'http': 'socks5h://127.0.0.1:9050',
    'https': 'socks5h://127.0.0.1:9050'
  }


##HTTPリクエスト成功失敗関わらず待機時間
  ##nitterでユーザーがいなかったとき返ってくるページのタイトル
##1秒待つだけで行儀がいいクローラーだそうなので既定では1秒
  ##万が一仕様変更で変わったとき用
##しかし日本のポリホーモは1秒待っていても捕まえてくるので注意
  NITTER_ERROR_TITLE: Final[str] = 'Error|nitter'
##https://ja.wikipedia.org/wiki/?curid=2187212
Waittime: Final[int] = 1


##nitterのURLのドメインとユーザーネーム部分の接続部品
  ##archive.todayで魚拓がなかったときのレスポンス
Searchquery: Final[str] = "search?q=from%3A"
  ##万が一仕様変更で変わったとき用
  NO_ARCHIVE: Final[str] = 'No results'


##HTTPリクエスト時のユーザーエージェントとヘッダ
  ##nitterで次ページ遷移時のURLから抜け落ちてる部分
user_agent: Final[str] = 'Mozilla/5.0 (Windows NT 10.0; rv:91.0) Gecko/20100101 Firefox/91.0'
  SEARCH: Final[str] = 'search'
header = {
    'User-Agent': user_agent
}


##nitterでユーザーがいなかったとき返ってくるページのタイトル
  ##nitterの前ページ読み込み部分の名前
##万が一仕様変更で変わったとき用
  ##万が一仕様変更で変わったとき用
Nitter_error_title: Final[str] = "Error|nitter"
  NEWEST: Final[str] = 'Load newest'


##archive.todayで魚拓がなかったときのレスポンス
  ##画像などのツイートメディアをダウンロードするディレクトリ
##万が一仕様変更で変わったとき用
  MEDIA_DIR: Final[str] = 'tweet_media'
Noarchive: Final[str] = "No results"


##nitterで次ページ遷移時のURLから抜け落ちてる部分
  #関数類
Search: Final[str] = "search"
  def __init__(self, krsw: bool=False):
    self.txt_data = ''
    self.limit_count = 0 ##記録数
    self._check_slash() ##スラッシュが抜けてないかチェック
    self._check_tor() ##Ｔorが使えているかチェック
    self._check_instance() ##インスタンスが死んでないかチェック


##nitterの前ページ読み込み部分の名前
    ##ユーザー名取得
##万が一仕様変更で変わったとき用
    if krsw:
Newest: Final[str] = "Load newest"
      print('名前は自動的に' + self.CALLINSHOW + 'にナリます')
      name: Final[str] = self.CALLINSHOW
    else:
      name: Final[str] = self._get_name()


#関数類
    ##検索クエリとページ取得
##坂根輝美に場所を知らせます
    if krsw:
def pickupcounter():
      print('クエリは自動的になしにナリます')
print('ピックアップカウンター付近でふ')
      self._page: str = self._request(self.NITTER_INSTANCE + self.SEARCH_QUERY + name)
      if self._page is None:
##引数のURLにHTTP接続します
        self._fail()
##失敗かどうかは呼出側で要判定
    else:
def request_onetime(url: Final[str]) -> requests.models.Response:
      self._page: str = self._get_query(name)
res =requests.get(url, timeout=Request_timeout, headers=header,allow_redirects=False)
sleep(Waittime) ##DOS対策で待つ
return res


##HTTP接続を再試行回数まで試します
    ##終わりにするツイート取得
##成功すると結果を返します
    if krsw:
##接続失敗が何度も起きるとNoneを返します
      print('終わりにするツイートは自動的になしにナリます')
##呼出側で要None判定
      self._stop: Final[str] = ''
def request(url: Final[str]) -> Optional[requests.models.Response]:
    else:
counter = 1 ##リクエスト挑戦回数を記録
      self._stop: Final[str] = self._stop_word()
while True:
try:
res = request_onetime(url) ##リクエスト
res.raise_for_status() ##HTTPステータスコードが200番台以外でエラー発生
except requests.exceptions.RequestException as e:
print(url + 'への通信失敗ナリ  ' + str(counter) + '/' + str(Limit_request) + '回')
if counter < Limit_request: ##エラー発生時上限まで再挑戦
counter += 1 ##現在の試行回数1回増やす
sleep(Waittime_error) ##失敗時は長めに待つ
else:
return None ##失敗したらNone返却し呼出側で対処してもらう
else:
return res ##リクエストの結果返す


##URLの最後にスラッシュ付いていなければ付ける
    print()
##Twitterだけスラッシュ付いていないほうが都合いいので抜く
def slash_check():
global Nitterinstance
global Archivetoday
global Archivetodaystandard
global Twitterurl
if Nitterinstance[-1] != '/':
Nitterinstance = Nitterinstance + '/'
if Archivetoday[-1] != '/':
Archivetoday = Archivetoday + '/'
if Archivetodaystandard[-1] != '/':
Archivetodaystandard = Archivetodaystandard + '/'
if Twitterurl[-1] == '/':
Twitterurl = Twitterurl[0:len(Twitterurl)-1]


##nitterのインスタンスが生きているかチェック
##死んでいたらそこで終了
##接続を一回しか試さないrequest_onetimeを使っているのは
##激重インスタンスが指定されたとき試行回数増やして偶然成功してそのまま実行されるのを躱すため
def instance_check():
print("nitterのインスタンスチェック中ですを")
try:
res = request_onetime(Nitterinstance) ##リクエスト
res.raise_for_status() ##HTTPステータスコードが200番台以外でエラー発生
except requests.exceptions.RequestException as e: ##エラー発生時は終了
print('インスタンスが死んでますを')
exit()


##ツイート収集するユーザー名を取得
  ##坂根輝美に場所を知らせます
##何も入力しないと尊師を指定するよう改良
  def _pickup_counter(self):
def get_name() -> str:
    print('ピックアップカウンター付近でふ')
while True:
print('アカウント名を入れなければない。空白だと自動的に' + Callinshow + 'になりますを')
account_str = input() ##ユーザー入力受付
##空欄で降臨ショー
if account_str == '':
return Callinshow
else:
res = request(Nitterinstance + account_str) ##リクエストして結果取得
if res is None : ##リクエスト失敗判定
fail()
soup = bs4(res.text, 'html.parser') ##beautifulsoupでレスポンス解析
if soup.title == Nitter_error_title: ##タイトルがエラーでないか判定
print(account_str + "は実在の人物ではありませんでした") ##エラー時ループに戻る
else:
print("最終的に出会ったのが@" + account_str + "だった。")
return account_str ##成功時アカウント名返す


##検索クエリを取得
  ##引数のURLにHTTP接続します
def get_query(name: Final[str]) -> requests.models.Response:
  ##失敗かどうかは呼出側で要判定
while True:
  def _request_once(self, url: Final[str]) -> requests.models.Response:
print("検索クエリを入れるナリ。複数ある時は一つの塊ごとに入力するナリ。空欄を入れると検索開始ナリ。")
    res = requests.get(url, timeout=self.REQUEST_TIMEOUT, headers=self.HEADERS, allow_redirects=False, proxies=self.PROXIES)
print("例：「無能」→改行→「脱糞」→改行→「弟殺し」→改行→空欄で改行")
    sleep(self.WAIT_TIME) ##DoS対策で待つ
query_str = [name] ##検索クエリ文字列をquery_strに収納。ユーザー名を先に含めておく
    return res
query_input = input() ##ユーザー入力受付
##空欄が押されるまでユーザー入力受付
while query_input != '':
query_str.append(quote(query_input))
query_input = input()
res = request(Nitterinstance + Searchquery + '+'.join(query_str)) ##リクエストして結果取得
if res is None : ##リクエスト失敗判定
fail()
soup = bs4(res.text, 'html.parser') ##beautifulsoupでレスポンス解析
##1件もないときはHTMLにtimeline-noneがあるから判定
if soup.find(class_='timeline-none') is None:
print("クエリのピースが埋まっていく。")
return res ##成功したリクエストのレスポンスを返す
else:
print("適切なクエリを入力することを切に望む。")


##接続失敗時処理
  ##HTTP接続を再試行回数まで試します
def fail():
  ##成功すると結果を返します
print("接続失敗しすぎで強制終了ナリ")
  ##接続失敗が何度も起きるとNoneを返します
if txt_data != '': ##取得成功したデータがあれば発行
  ##呼出側で要None判定
print("取得成功した分だけ発行しますを")
  def _request(self, url: Final[str]) -> Optional[requests.models.Response]:
make_txt()
    counter = 1 ##リクエスト挑戦回数を記録
exit() ##終了
    while True:
      try:
        res = self._request_once(url) ##リクエスト
        res.raise_for_status() ##HTTPステータスコードが200番台以外でエラー発生
      except requests.exceptions.RequestException as e:
        print(url + 'への通信失敗ナリ  ' + f"{counter}/{self.LIMIT_N_REQUESTS}回")
        if counter < self.LIMIT_N_REQUESTS: ##エラー発生時上限まで再挑戦
          counter += 1 ##現在の試行回数1回増やす
          sleep(self.WAIT_TIME_FOR_ERROR) ##失敗時は長めに待つ
        else:
          return None ##失敗したらNone返却し呼出側で対処してもらう
      else:
        return res ##リクエストの結果返す


##テキスト発行
  ##URLの最後にスラッシュ付いていなければ付ける
def make_txt():
  ##Twitterだけスラッシュ付いていないほうが都合いいので抜く
global txt_data
  def _check_slash(self) -> Union[None, NoReturn]:
txt_data = '{|class="wikitable" style="text-align: left;"\n' + txt_data + '|}' ##wikiの表の最初と最後
    if self.NITTER_INSTANCE[-1] != '/':
##ファイル出力
      raise RuntimeError('NITTER_INSTANCEの末尾には/が必須です')
with codecs.open('tweet.txt', 'w', 'utf-8') as f:
    if self.ARCHIVE_TODAY[-1] != '/':
f.write(txt_data)
      raise RuntimeError('ARCHIVE_TODAYの末尾には/が必須です')
print("テキストファイル手に入ったやで〜")
    if self.ARCHIVE_TODAY_STANDARD[-1] != '/':
exit() ##終了
      raise RuntimeError('ARCHIVE_TODAY_STANDARDの末尾には/が必須です')
    if self.TWITTER_URL[-1] == '/':
      raise RuntimeError('TWITTER_URLの末尾には/があってはいけません')


##記録を中断するツイート
  ##Ｔorが使えているかチェック
def stop_word() -> str:
  def _check_tor(self) -> Union[None, NoReturn]:
print("ここにツイートの本文を入れる！すると・・・・なんと一致するツイートで記録を中断する！（これは本当の仕様です。）ちなみに、空欄だと" + str(Limit_tweet) + "件まで終了しない。")
    print('Ｔorのチェック中ですを')
end_str = input() ##ユーザー入力受付
    res = self._request_once('https://check.torproject.org/api/ip') ##リクエスト
return end_str
    is_tor = json.loads(res.text)['IsTor']
    if is_tor:
      print('Ｔor OK')
    else:
      raise RuntimeError('通信がＴorを経由していないなりを')


#ページからツイート本文をtxt_dataに収めていく
  ##nitterのインスタンスが生きているかチェック
def get_tweet(page: Final[requests.models.Response],stop: Final[str]):
  ##死んでいたらそこで終了
global txt_data
  ##接続を一回しか試さない_request_onceを使っているのは
global limitcount
  ##激重インスタンスが指定されたとき試行回数増やして偶然成功してそのまま実行されるのを躱すため
soup = bs4(page.text, 'html.parser') ##beautifulsoupでレスポンス解析
  def _check_instance(self) -> Union[None, NoReturn]:
tweets = soup.find_all(class_='timeline-item') ##一ツイートのブロックごとにリストで取得
    print("nitterのインスタンスチェック中ですを")
for tweet in tweets: ##一ツイート毎に処理
    try:
doublesoup = bs4(str(tweet) , 'html.parser') ##ツイートをさらに解析
      res = self._request_once(self.NITTER_INSTANCE) ##リクエスト
if doublesoup.a.text == Newest: ##Load Newestのボタンは処理しない
      res.raise_for_status() ##HTTPステータスコードが200番台以外でエラー発生
continue
    except requests.exceptions.RequestException as e: ##エラー発生時は終了
if not doublesoup.find(class_='retweet-header') is None: ##retweet-headerはリツイートを示すので入っていれば処理しない
      print('インスタンスが死んでますを')
continue
      exit()
tweet_url = Twitterurl + re.sub('#[^#]*$','',doublesoup.find(class_='tweet-link').get('href')) ##ツイートのURL作成
archived_tweet_url = callinshowlinkurl(tweet_url) ##ツイートURLをテンプレートCallinShowlinkに変化
tweet_content = doublesoup.find(class_='tweet-content media-body') ##ツイートの中身だけ取り出す
triplesoup = bs4(str(tweet_content) , 'html.parser') ##URLの魚拓化のためにさらに解析
archivesoup(triplesoup) ##ツイートの中身のリンクをテンプレートArchiveに変化
txt_data = '!' + archived_tweet_url + '\n|-\n|\n' + triplesoup.get_text().replace('\n','<br>\n').replace('#','<nowiki>#</nowiki>') + '\n|-\n' + txt_data ##wikiの文法に変化
limitcount += 1 ##記録回数をカウント
print("ツイートを" + str(limitcount) + "件も記録したンゴｗｗｗｗｗｗｗｗｗｗｗ")
if stop != '' and stop in triplesoup.get_text(): ##目的ツイートか判定
print("目的ツイート発見でもう尾張屋根")
make_txt()
if limitcount >= Limit_tweet: ##上限達成か判定
print(str(Limit_tweet) + "件も記録している。もうやめにしませんか。")
make_txt()


#soupをテンプレートArchiveに変化させる
  ##ツイート収集するユーザー名を取得
def archivesoup(soup):
  ##何も入力しないと尊師を指定するよう改良
urls_in_tweet = soup.find_all('a')
  def _get_name(self) -> Union[str, NoReturn]:
for url in urls_in_tweet:
    while True:
if not re.match('^https?://',url.get('href')) is None: ##先頭にhttpが付いていない物はハッシュタグの検索ページへのリンクなので処理しない
      print('アカウント名を入れなければない。空白だと自動的に' + self.CALLINSHOW + 'になりますを')
newstr = soup.new_string(archiveurl(url.get('href'),url.text)) ##テンプレートArchiveの文字列作成
      account_str = input() ##ユーザー入力受付
url.replace_with(newstr) ##テンプレートArchiveに変化
      ##空欄で降臨ショー
      if account_str == '':
        return self.CALLINSHOW
      else:
        res = self._request(self.NITTER_INSTANCE + account_str) ##リクエストして結果取得
        if res is None : ##リクエスト失敗判定
          self._fail()
        soup = BeautifulSoup(res.text, 'html.parser') ##beautifulsoupでレスポンス解析
        if soup.title == self.NITTER_ERROR_TITLE: ##タイトルがエラーでないか判定
          print(account_str + "は実在の人物ではありませんでした") ##エラー時ループに戻る
        else:
          print("最終的に出会ったのが@" + account_str + "だった。")
          return account_str ##成功時アカウント名返す


#URLをテンプレートArchiveに変化させる
  ##検索クエリを取得
def archiveurl(url: Final[str],text: Final[str]) -> str:
  def _get_query(self, name: Final[str]) -> Union[requests.models.Response, NoReturn]:
return '{{Archive|1=' + url + '|2=' + archive(url) + '|3=' + text  + '}}' ##テンプレートArchiveの文字列返す
    while True:
      print("検索クエリを入れるナリ。複数ある時は一つの塊ごとに入力するナリ。空欄を入れると検索開始ナリ。")
#URLをテンプレートCallinShowlinkに変化させる
      print("例：「無能」→改行→「脱糞」→改行→「弟殺し」→改行→空欄で改行")
def callinshowlinkurl(url: Final[str]) -> str:
      query_str = [name] ##検索クエリ文字列をquery_strに収納。ユーザー名を先に含めておく
return '{{CallinShowLink|1=' + url + '|2=' + archive(url) + '}}' ##テンプレートCallinShowlinkの文字列返す
      query_input = input() ##ユーザー入力受付
      ##空欄が押されるまでユーザー入力受付
      while query_input != '':
        query_str.append(quote(query_input))
        query_input = input()
      res = self._request(self.NITTER_INSTANCE + self.SEARCH_QUERY + '+'.join(query_str)) ##リクエストして結果取得
      if res is None : ##リクエスト失敗判定
        self._fail()
      soup = BeautifulSoup(res.text, 'html.parser') ##beautifulsoupでレスポンス解析
      ##1件もないときはHTMLにtimeline-noneがあるから判定
      if soup.find(class_='timeline-none') is None:
        print("クエリのピースが埋まっていく。")
        return res ##成功したリクエストのレスポンスを返す
      else:
        print("適切なクエリを入力することを切に望む。")


##URLから魚拓返す
  ##接続失敗時処理
def archive(url: Final[str]) -> str:
  def _fail(self) -> NoReturn:
archive_url = Archivetodaystandard + url ##wikiに載せるとき用URLで失敗するとこのままhttps://archive.ph/https://xxxxxxxxの形で返される
    print("接続失敗しすぎで強制終了ナリ")
res = request(Archivetoday + url) ##アクセス用URL使って結果を取得
    if self.txt_data != '': ##取得成功したデータがあれば発行
if res is None : ##魚拓接続失敗時処理
      print("取得成功した分だけ発行しますを")
print(archive_url + 'にアクセス失敗ナリ。出力されるテキストにはそのまま記載されるナリ。')
      self._make_txt()
else:
    exit() ##終了
soup = bs4(res.text, 'html.parser') ##beautifulsoupでレスポンス解析
content = soup.find(id="CONTENT") ##archive.todayの魚拓一覧ページの中身だけ取得
if content.get_text()[:len(Noarchive)] == Noarchive: ##魚拓があるかないか判定
print(url + "の魚拓がない。これはいけない。")
else:
doublesoup = bs4(str(content), 'html.parser') ##beautifulsoupでレスポンス解析
archive_url = doublesoup.find('a').get('href').replace(Archivetoday,Archivetodaystandard)
return archive_url


##新しいページを取得
  ##テキスト発行
def getnewpage(page: Final[requests.models.Response]) -> requests.models.Response:
  def _make_txt(self) -> NoReturn:
soup = bs4(page.text, 'html.parser') ##beautifulsoupでレスポンス解析
    self.txt_data = '{|class="wikitable" style="text-align: left;"\n' + self.txt_data + '|}' ##wikiの表の最初と最後
showmores = soup.find_all(class_="show-more")
    ##ファイル出力
for showmore in showmores: ##show-moreに次ページへのリンクか前ページへのリンクがある
    with codecs.open('tweet.txt', 'w', 'utf-8') as f:
if showmore.text != Newest:  ##前ページへのリンクではないか判定
      f.write(self.txt_data)
newurl = Nitterinstance + Search + showmore.a.get('href') ##直下のaタグのhrefの中身取ってURL頭部分と合体
    print("テキストファイル手に入ったやで〜")
res = request(newurl) ##接続してHTML取ってくる
    exit() ##終了
if res is None:
fail()
newsoup = bs4(res.text, 'html.parser') ##beautifulsoupでレスポンス解析
if newsoup.find(class_="timeline-end") is None: ##ツイートの終端ではtimeline-endだけのページになるので判定
print(res.url + 'に移動しますを')
return res ##まだ残りツイートがあるのでページを返して再度ツイート本文収集
else:
print("急に残りツイートが無くなったな終了するか")
make_txt()


  ##記録を中断するツイート
  def _stop_word(self) -> str:
    print(f"ここにツイートの本文を入れる！すると・・・・なんと一致するツイートで記録を中断する！（これは本当の仕様です。）ちなみに、空欄だと{self.LIMIT_N_TWEETS}件まで終了しない。")
    end_str = input() ##ユーザー入力受付
    return end_str


  def _mkdir(self):
    try:
      os.mkdir(self.MEDIA_DIR)
    except FileExistsError:
      pass


##メイン
  ##Twitterの画像をダウンロード
txt_data = '' ##出力するデータ
  def _download_media(self, media_name) -> bool:
limitcount = 0 ##記録数
    self._mkdir()
krsw=False ##コマンドライン引数があるかどうかのフラグ
    url = 'https://pbs.twimg.com/media/' + media_name
##コマンドライン引数取得
    res = self._request(url)
if len(sys.argv) > 1 and sys.argv[1] == 'krsw':
    if res is not None:
krsw=True
      if 'image' not in res.headers['content-type']:
slash_check() ##スラッシュが抜けてないかチェック
        return False
instance_check() ##インスタンスが死んでないかチェック
      with open(self.MEDIA_DIR + '/' + media_name, "wb") as f:
        f.write(res.content)
      return True
    else:
      return False


##ユーザー名取得
  def _get_tweet_media(self, tweet: bs4.element.Tag) -> str:
if krsw:
    tweet_media = tweet.select_one('.tweet-body > .attachments') # 引用リツイート内のメディアを選択しないように.tweet-body直下の.attachmentsを選択
print('名前は自動的に' + Callinshow + 'にナリます')
    media_txt = ''
name = Callinshow
    if tweet_media is not None:
else:
      for tweet_medium in tweet_media:
name = get_name()
        # ツイートの画像の取得
        if tweet_medium.select_one('.attachment.image a'):
          media_name = re.search(r'%2F([^%]*\.jpg)|%2F([^%]*\.jpeg)|%2F([^%]*\.png)|%2F([^%]*\.gif)', tweet_medium.select_one('.attachment.image a').get('href')).group(1)
          media_txt += f"<br>\n[[ファイル:{media_name}|240px]]"
          if self._download_media(media_name):
            print(self.MEDIA_DIR + '/' + media_name + ' をアップロードしなければない。')
          else:
            print('https://pbs.twimg.com/media/' + media_name + ' をアップロードしなければない。')
        # 動画についてはm3u8で落ちてきて面倒臭いため取得しない
        """
        elif tweet_medium.select_one('.attachment.video-container video'):
          media_url = unquote(re.search(r'[^\/]+$', tweet_medium.select_one('.attachment.video-container video').get('data-url')).group(0))
          print(media_url)
        """
    return media_txt


##検索クエリとページ取得
  def _get_tweet_quote(self, tweet: bs4.element.Tag) -> str:
if krsw:
    tweet_quote = tweet.select_one('.tweet-body > .quote.quote-big') # 引用リツイートを選択
print('クエリは自動的になしにナリます')
    quote_txt = ''
page = request(Nitterinstance + Searchquery + name)
    if tweet_quote is not None:
if page is None:
      link = tweet_quote.select_one('.quote-link').get('href')
fail()
      link = re.sub('#.*$', '', link)
else:
      link = self.TWITTER_URL + link
page = get_query(name)
      quote_txt = '<br>\n' + self._archive_url(link, link)
    return quote_txt


##終わりにするツイート取得
  #ページからツイート本文をself.txt_dataに収めていく
if krsw:
  def get_tweet(self) -> Union[None, NoReturn]:
print('終わりにするツイートは自動的になしにナリます')
    soup = BeautifulSoup(self._page.text, 'html.parser') ##beautifulsoupでレスポンス解析
stop = ''
    tweets = soup.find_all(class_='timeline-item') ##一ツイートのブロックごとにリストで取得
else:
    for tweet in tweets: ##一ツイート毎に処理
stop = stop_word()
      if tweet.a.text == self.NEWEST: ##Load Newestのボタンは処理しない
        continue
      if tweet.find(class_='retweet-header') is not None: ##retweet-headerはリツイートを示すので入っていれば処理しない
        continue
      tweet_url = self.TWITTER_URL + re.sub('#[^#]*$', '', tweet.find(class_='tweet-link').get('href')) ##ツイートのURL作成
      archived_tweet_url = self._callinshowlink_url(tweet_url) ##ツイートURLをテンプレートCallinShowlinkに変化
      tweet_content = tweet.find(class_='tweet-content media-body') ##ツイートの中身だけ取り出す
      tweet_content = self._archive_soup(tweet_content) ##ツイートの中身のリンクをテンプレートArchiveに変化
      media_txt = self._get_tweet_media(tweet) ##ツイートに画像などのメディアを追加
      quote_txt = self._get_tweet_quote(tweet) ##引用リツイートの場合、元ツイートを追加
      self.txt_data = '!' + archived_tweet_url + '\n|-\n|\n' + tweet_content.get_text().replace('\n', '<br>\n').replace('#', '<nowiki>#</nowiki>') + quote_txt + media_txt + '\n|-\n' + self.txt_data ##wikiの文法に変化
      self.limit_count += 1 ##記録回数をカウント
      if self.limit_count % self.REPORT_INTERVAL == 0:
        print(f"ツイートを{self.limit_count}件も記録したンゴｗｗｗｗｗｗｗｗｗｗｗ")
      if self._stop != '' and self._stop in tweet_content.get_text(): ##目的ツイートか判定
        print("目的ツイート発見でもう尾張屋根")
        self._make_txt()
      if self.limit_count >= self.LIMIT_N_TWEETS: ##上限達成か判定
        print(f"{self.LIMIT_N_TWEETS}件も記録している。もうやめにしませんか。")
        self._make_txt()


##ツイートを取得し終えるまでループ
  #soupをテンプレートArchiveに変化させる
while True:
  def _archive_soup(self, tag: bs4.element.Tag) -> BeautifulSoup:
get_tweet(page,stop) ##txt_dataにページ内のリツイート以外の全ツイートの中身突っ込んでいく
    soup = BeautifulSoup(str(tag))
page = getnewpage(page) ##新しいページ取得
    urls_in_tweet = soup.find_all('a')
    for url in urls_in_tweet:
      if re.match('^https?://', url.get('href')) is not None: ##先頭にhttpが付いていない物はハッシュタグの検索ページへのリンクなので処理しない
        newstr = soup.new_string(self._archive_url(url.get('href'), url.text)) ##テンプレートArchiveの文字列作成
        url.replace_with(newstr) ##テンプレートArchiveに変化
    return soup
 
  #URLをテンプレートArchiveに変化させる
  def _archive_url(self, url: Final[str], text: Final[str]) -> str:
    return '{{Archive|1=' + url + '|2=' + self._archive(url) + '|3=' + text  + '}}' ##テンプレートArchiveの文字列返す
 
  #URLをテンプレートCallinShowlinkに変化させる
  def _callinshowlink_url(self, url: Final[str]) -> str:
    return '{{CallinShowLink|1=' + url + '|2=' + self._archive(url) + '}}' ##テンプレートCallinShowlinkの文字列返す
 
  ##URLから魚拓返す
  def _archive(self, url: Final[str]) -> str:
    archive_url = self.ARCHIVE_TODAY_STANDARD + url ##wikiに載せるとき用URLで失敗するとこのままhttps://archive.ph/https://xxxxxxxxの形で返される
    res = self._request(self.ARCHIVE_TODAY + url) ##アクセス用URL使って結果を取得
    if res is None : ##魚拓接続失敗時処理
      print(archive_url + 'にアクセス失敗ナリ。出力されるテキストにはそのまま記載されるナリ。')
    else:
      soup = BeautifulSoup(res.text, 'html.parser') ##beautifulsoupでレスポンス解析
      content = soup.find(id="CONTENT") ##archive.todayの魚拓一覧ページの中身だけ取得
      if content.get_text()[:len(self.NO_ARCHIVE)] == self.NO_ARCHIVE: ##魚拓があるかないか判定
        print(url + "の魚拓がない。これはいけない。")
      else:
        archive_url = content.find('a').get('href').replace(self.ARCHIVE_TODAY, self.ARCHIVE_TODAY_STANDARD)
    return archive_url
 
  ##新しいページを取得
  def go_to_new_page(self) -> Union[None, NoReturn]:
    soup = BeautifulSoup(self._page.text, 'html.parser') ##beautifulsoupでレスポンス解析
    showmores = soup.find_all(class_="show-more")
    for showmore in showmores: ##show-moreに次ページへのリンクか前ページへのリンクがある
      if showmore.text != self.NEWEST:  ##前ページへのリンクではないか判定
        newurl = self.NITTER_INSTANCE + self.SEARCH + showmore.a.get('href') ##直下のaタグのhrefの中身取ってURL頭部分と合体
    res = self._request(newurl) ##接続してHTML取ってくる
    if res is None:
      self._fail()
    newsoup = BeautifulSoup(res.text, 'html.parser') ##beautifulsoupでレスポンス解析
    if newsoup.find(class_="timeline-end") is None: ##ツイートの終端ではtimeline-endだけのページになるので判定
      print(res.url + 'に移動しますを')
      self._page = res ##まだ残りツイートがあるのでページを返して再度ツイート本文収集
    else:
      print("急に残りツイートが無くなったな終了するか")
      self._make_txt()
 
 
if __name__ == '__main__':
  krsw = len(sys.argv) > 1 and sys.argv[1] == 'krsw' ##コマンドライン引数があるかどうかのフラグ
  twitter_archiver = TwitterArchiver(krsw)
 
  ##ツイートを取得し終えるまでループ
  while True:
    twitter_archiver.get_tweet() ##self.txt_dataにページ内のリツイート以外の全ツイートの中身突っ込んでいく
    twitter_archiver.go_to_new_page() ##新しいページ取得
‎</syntaxhighlight>
‎</syntaxhighlight>
== 実行例 ==
== 実行例 ==
20件での実行例。
20件での実行例。
匿名利用者
「https://krsw-wiki.in/wiki/特別:携帯機器差分/94638」から取得

案内メニュー