利用者:夜泣き/スクリプト
< 利用者:夜泣き
とりあえず取り急ぎ。バグ報告は利用者・トーク:夜泣き
コード
#!/usr/bin/env python3
'''
当コードは恒心停止してしまったhttps://rentry.co/7298gの降臨ショーツイート自動収集スクリプトの復刻改善版です
前開発者との出会いに感謝
〜〜〜〜〜〜〜〜〜〜〜〜〜【使い方】〜〜〜〜〜〜〜〜〜〜〜〜〜
・terminalに $ python3 (ファイル名) で作動します
・定数類は状況に応じて変えてください
・$ python3 (ファイル名) krsw コマンドライン引数をkrswとつけると自動モードです
・自動モードではユーザーは降臨ショー、クエリはなし、取得上限まで自動です
・つまりユーザー入力が要りません
〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜
ーーーーーーーーーーーーー【注意点】ーーーーーーーーーーーーー
・環境は玉葱前提です。
・Whonix-Workstationで動作確認済
・bs4はインストールしないと標準で入ってません
・requestsも環境によっては入っていないかもしれない
・$ pip install bs4 requests
・pipも入っていなければ$ sudo apt install pip
・バグ報告はhttps://krsw-wiki.org/wiki/?curid=15799
ーーーーーーーーーーーーーーーーーーーーーーーーーーーーーーー
'''
#インポート類
import sys
import codecs
import re
from time import sleep
import requests
from bs4 import BeautifulSoup as bs4
from typing import Final
from typing import Optional
from urllib.parse import quote
import warnings
#定数・設定類
##おそらくはツイートの内容によってMarkupResemblesLocatorWarningが吐き出されることがあるので無効化
warnings.simplefilter('ignore')
##nitterのインスタンス
##生きているのはhttps://github.com/zedeus/nitter/wiki/Instancesで確認
Nitterinstance = 'http://nitterqdyumlovt7tjqpdjrluitgmtpa53qq3idlpgoe4kxo7gs3xvad.onion/'
##archive.todayの魚拓
##実際にアクセスして魚拓があるか調べるのにはonionを使用
Archivetoday = 'http://archiveiya74codqgiixo33q62qlrqtkgmcitqx5u2oeqnmn5bpcbiyd.onion/'
##archive.todayの魚拓
##記事の文章に使うのはクリアネット
Archivetodaystandard = 'https://archive.ph/'
##twitterのURL
Twitterurl = 'https://twitter.com/'
##降臨ショーのユーザーネーム
Callinshow: Final[str] = 'CallinShow'
##HTTPリクエストのタイムアウト秒数
Request_timeout: Final[int] = 30
##取得するツイート数の上限
Limit_tweet: Final[int] = 100
##HTTPリクエスト失敗時の再試行回数
Limit_request: Final[int] = 5
##HTTPリクエスト失敗時にさらに追加する待機時間
Waittime_error: Final[int] = 4
##HTTPリクエスト成功失敗関わらず待機時間
##1秒待つだけで行儀がいいクローラーだそうなので既定では1秒
##しかし日本のポリホーモは1秒待っていても捕まえてくるので注意
##https://ja.wikipedia.org/wiki/?curid=2187212
Waittime: Final[int] = 1
##nitterのURLのドメインとユーザーネーム部分の接続部品
Searchquery: Final[str] = "search?q=from%3A"
##HTTPリクエスト時のユーザーエージェントとヘッダ
user_agent: Final[str] = 'Mozilla/5.0 (Windows NT 10.0; rv:91.0) Gecko/20100101 Firefox/91.0'
header = {
'User-Agent': user_agent
}
##nitterでユーザーがいなかったとき返ってくるページのタイトル
##万が一仕様変更で変わったとき用
Nitter_error_title: Final[str] = "Error|nitter"
##archive.todayで魚拓がなかったときのレスポンス
##万が一仕様変更で変わったとき用
Noarchive: Final[str] = "No results"
##nitterで次ページ遷移時のURLから抜け落ちてる部分
Search: Final[str] = "search"
##nitterの前ページ読み込み部分の名前
##万が一仕様変更で変わったとき用
Newest: Final[str] = "Load newest"
#関数類
##坂根輝美に場所を知らせます
def pickupcounter():
print('ピックアップカウンター付近でふ')
##引数のURLにHTTP接続します
##失敗かどうかは呼出側で要判定
def request_onetime(url: Final[str]) -> requests.models.Response:
res =requests.get(url, timeout=Request_timeout, headers=header,allow_redirects=False)
sleep(Waittime) ##DOS対策で待つ
return res
##HTTP接続を再試行回数まで試します
##成功すると結果を返します
##接続失敗が何度も起きるとNoneを返します
##呼出側で要None判定
def request(url: Final[str]) -> Optional[requests.models.Response]:
counter = 1 ##リクエスト挑戦回数を記録
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の最後にスラッシュ付いていなければ付ける
##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 get_name() -> str:
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 ##成功時アカウント名返す
##検索クエリを取得
def get_query(name: Final[str]) -> requests.models.Response:
while True:
print("検索クエリを入れるナリ。複数ある時は一つの塊ごとに入力するナリ。空欄を入れると検索開始ナリ。")
print("例：「無能」→改行→「脱糞」→改行→「弟殺し」→改行→空欄で改行")
query_str = [name] ##検索クエリ文字列をquery_strに収納。ユーザー名を先に含めておく
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("適切なクエリを入力することを切に望む。")
##接続失敗時処理
def fail():
print("接続失敗しすぎで強制終了ナリ")
if txt_data != '': ##取得成功したデータがあれば発行
print("取得成功した分だけ発行しますを")
make_txt()
exit() ##終了
##テキスト発行
def make_txt():
global txt_data
txt_data = '{|class="wikitable" style="text-align: left;"\n' + txt_data + '|}' ##wikiの表の最初と最後
##ファイル出力
with codecs.open('tweet.txt', 'w', 'utf-8') as f:
f.write(txt_data)
print("テキストファイル手に入ったやで〜")
exit() ##終了
##記録を中断するツイート
def stop_word() -> str:
print("ここにツイートの本文を入れる！すると・・・・なんと一致するツイートで記録を中断する！（これは本当の仕様です。）ちなみに、空欄だと" + str(Limit_tweet) + "件まで終了しない。")
end_str = input() ##ユーザー入力受付
return end_str
#ページからツイート本文をtxt_dataに収めていく
def get_tweet(page: Final[requests.models.Response],stop: Final[str]):
global txt_data
global limitcount
soup = bs4(page.text, 'html.parser') ##beautifulsoupでレスポンス解析
tweets = soup.find_all(class_='timeline-item') ##一ツイートのブロックごとにリストで取得
for tweet in tweets: ##一ツイート毎に処理
doublesoup = bs4(str(tweet) , 'html.parser') ##ツイートをさらに解析
if doublesoup.a.text == Newest: ##Load Newestのボタンは処理しない
continue
if not doublesoup.find(class_='retweet-header') is None: ##retweet-headerはリツイートを示すので入っていれば処理しない
continue
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')
for url in urls_in_tweet:
if not re.match('^https?://',url.get('href')) is None: ##先頭にhttpが付いていない物はハッシュタグの検索ページへのリンクなので処理しない
newstr = soup.new_string(archiveurl(url.get('href'),url.text)) ##テンプレートArchiveの文字列作成
url.replace_with(newstr) ##テンプレートArchiveに変化
#URLをテンプレートArchiveに変化させる
def archiveurl(url: Final[str],text: Final[str]) -> str:
return '{{Archive|1=' + url + '|2=' + archive(url) + '|3=' + text + '}}' ##テンプレートArchiveの文字列返す
#URLをテンプレートCallinShowlinkに変化させる
def callinshowlinkurl(url: Final[str]) -> str:
return '{{CallinShowLink|1=' + url + '|2=' + archive(url) + '}}' ##テンプレートCallinShowlinkの文字列返す
##URLから魚拓返す
def archive(url: Final[str]) -> str:
archive_url = Archivetodaystandard + url ##wikiに載せるとき用URLで失敗するとこのままhttps://archive.ph/https://xxxxxxxxの形で返される
res = request(Archivetoday + url) ##アクセス用URL使って結果を取得
if res is None : ##魚拓接続失敗時処理
print(archive_url + 'にアクセス失敗ナリ。出力されるテキストにはそのまま記載されるナリ。')
else:
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:
soup = bs4(page.text, 'html.parser') ##beautifulsoupでレスポンス解析
showmores = soup.find_all(class_="show-more")
for showmore in showmores: ##show-moreに次ページへのリンクか前ページへのリンクがある
if showmore.text != Newest: ##前ページへのリンクではないか判定
newurl = Nitterinstance + Search + showmore.a.get('href') ##直下のaタグのhrefの中身取ってURL頭部分と合体
res = request(newurl) ##接続してHTML取ってくる
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()
##メイン
txt_data = '' ##出力するデータ
limitcount = 0 ##記録数
krsw=False ##コマンドライン引数があるかどうかのフラグ
##コマンドライン引数取得
if len(sys.argv) > 1 and sys.argv[1] == 'krsw':
krsw=True
slash_check() ##スラッシュが抜けてないかチェック
instance_check() ##インスタンスが死んでないかチェック
##ユーザー名取得
if krsw:
print('名前は自動的に' + Callinshow + 'にナリます')
name = Callinshow
else:
name = get_name()
##検索クエリとページ取得
if krsw:
print('クエリは自動的になしにナリます')
page = request(Nitterinstance + Searchquery + name)
if page is None:
fail()
else:
page = get_query(name)
##終わりにするツイート取得
if krsw:
print('終わりにするツイートは自動的になしにナリます')
stop = ''
else:
stop = stop_word()
##ツイートを取得し終えるまでループ
while True:
get_tweet(page,stop) ##txt_dataにページ内のリツイート以外の全ツイートの中身突っ込んでいく
page = getnewpage(page) ##新しいページ取得
実行例
20件での実行例。