利用者:夜泣き/スクリプト
< 利用者:夜泣き
ナビゲーションに移動
検索に移動
とりあえず取り急ぎ。バグ報告は利用者・トーク:夜泣き
コード
#!/usr/bin/env python3
'''
ver2.2.7 2023/3/5恒心
当コードは恒心停止してしまったhttps://rentry.co/7298gの降臨ショーツイート自動収集スクリプトの復刻改善版です
前開発者との出会いに感謝
〜〜〜〜〜〜〜〜〜〜〜〜〜【使い方】〜〜〜〜〜〜〜〜〜〜〜〜〜
・terminalに $ python3 (ファイル名) で作動します
・定数類は状況に応じて変えてください
・$ python3 (ファイル名) krsw コマンドライン引数をkrswとつけると自動モードです
・自動モードではユーザーは降臨ショー、クエリはなし、取得上限まで自動です
・つまりユーザー入力が要りません
・ffmpegが入っていると動画も自動取得しますが、無くても動きます
〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜〜
ーーーーーーーーーーーーー【注意点】ーーーーーーーーーーーーー
・Pythonのバージョンは3.10以上
・環境は玉葱前提です。
・Whonix-Workstation, MacOSで動作確認済
・MacOSの場合はbrewでtorコマンドを導入し、実行
・PySocks, bs4はインストールしないと標準で入ってません
・requestsも環境によっては入っていないかもしれない
・$ pip install bs4 requests PySocks
・pipも入っていなければ$ sudo apt install pip
・バグ報告はhttps://krsw-wiki.org/wiki/?curid=15799#利用者:夜泣き/スクリプトについて
ーーーーーーーーーーーーーーーーーーーーーーーーーーーーーーー
'''
#インポート類
import sys
import os
import codecs
import re
import json
from datetime import datetime
from zoneinfo import ZoneInfo
from time import sleep
from typing import Final, NoReturn, TypeAlias
from urllib.parse import quote, unquote, urljoin
import warnings
import subprocess
import requests
import bs4
from bs4 import BeautifulSoup
##おそらくはツイートの内容によってMarkupResemblesLocatorWarningが吐き出されることがあるので無効化
warnings.simplefilter('ignore')
##型エイリアス
Response: TypeAlias = requests.models.Response
class TwitterArchiver:
#定数・設定類
##nitterのインスタンス
##生きているのはhttps://github.com/zedeus/nitter/wiki/Instancesで確認
##末尾にスラッシュ必須
NITTER_INSTANCE: Final[str] = 'http://nitter7bryz3jv7e3uekphigvmoyoem4al3fynerxkj22dmoxoq553qd.onion/'
##archive.todayの魚拓
##実際にアクセスして魚拓があるか調べるのにはonionを使用
##末尾にスラッシュ必須
ARCHIVE_TODAY: Final[str] = 'http://archiveiya74codqgiixo33q62qlrqtkgmcitqx5u2oeqnmn5bpcbiyd.onion/'
##archive.todayの魚拓
##記事の文章に使うのはクリアネット
##末尾にスラッシュ必須
ARCHIVE_TODAY_STANDARD: Final[str] = 'https://archive.vn/'
##twitterのURL
##末尾にスラッシュ必須
TWITTER_URL: Final[str] = 'https://twitter.com/'
##降臨ショーのユーザーネーム
CALLINSHOW: Final[str] = 'CallinShow'
##HTTPリクエストのタイムアウト秒数
REQUEST_TIMEOUT: Final[int] = 30
##取得するツイート数の上限
LIMIT_N_TWEETS: Final[int] = 100
##記録件数を報告するインターバル
REPORT_INTERVAL: Final[int] = 5
##HTTPリクエスト失敗時の再試行回数
LIMIT_N_REQUESTS: Final[int] = 5
##HTTPリクエスト失敗時にさらに追加する待機時間
WAIT_TIME_FOR_ERROR: Final[int] = 4
##HTTPリクエスト成功失敗関わらず待機時間
##1秒待つだけで行儀がいいクローラーだそうなので既定では1秒
##しかし日本のポリホーモは1秒待っていても捕まえてくるので注意
##https://ja.wikipedia.org/wiki/?curid=2187212
WAIT_TIME: Final[int] = 1
##nitterのURLのドメインとユーザーネーム部分の接続部品
SEARCH_QUERY: Final[str] = 'search?q=from%3A'
##HTTPリクエスト時のユーザーエージェントとヘッダ
HEADERS: Final[dict[str, str]] = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; rv:91.0) Gecko/20100101 Firefox/91.0'
}
##HTTP
PROXIES: Final[dict[str, str]] = {
'http': 'socks5h://127.0.0.1:9050',
'https': 'socks5h://127.0.0.1:9050'
}
##nitterでユーザーがいなかったとき返ってくるページのタイトル
##万が一仕様変更で変わったとき用
NITTER_ERROR_TITLE: Final[str] = 'Error|nitter'
##archive.todayで魚拓がなかったときのレスポンス
##万が一仕様変更で変わったとき用
NO_ARCHIVE: Final[str] = 'No results'
##nitterで次ページ遷移時のURLから抜け落ちてる部分
SEARCH: Final[str] = 'search'
##nitterの前ページ読み込み部分の名前
##万が一仕様変更で変わったとき用
NEWEST: Final[str] = 'Load newest'
##画像などのツイートメディアをダウンロードするディレクトリ
MEDIA_DIR: Final[str] = 'tweet_media'
#関数類
def __init__(self, krsw: bool=False):
self._txt_data: list[str] = []
self._limit_count: int = 0 ##記録数
self._proxy_is_needed: bool = False ## アクセスにプロキシが必要かどうか
self._check_slash() ##スラッシュが抜けてないかチェック
self._proxy_is_needed = self._check_tor_proxy_is_needed() ##Ｔorが使えているかチェック
self._check_nitter_instance() ##インスタンスが死んでないかチェック
self._check_archive_instance()
##Invidiousのインスタンスリストの正規表現パターンを取得
invidious_url_tuple: Final[tuple[str]] = self._invidious_instances()
self._invidious_pattern: Final[re.Pattern] = re.compile('|'.join(invidious_url_tuple))
##ユーザー名取得
if krsw:
print('名前は自動的に' + self.CALLINSHOW + 'にナリます')
name: Final[str] = self.CALLINSHOW
else:
name: Final[str] = self._get_name()
##検索クエリとページ取得
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)
##終わりにするツイート取得
if krsw:
print('終わりにするツイートは自動的になしにナリます')
self._stop: Final[str] = ''
else:
self._stop: Final[str] = self._stop_word()
##日付取得
self._date: datetime = self._tweet_date(BeautifulSoup(self._page.text, 'html.parser').find(class_='timeline-item'))
self._txt_data.append('')
print()
##坂根輝美に場所を知らせます
def _pickup_counter(self) -> None:
print('ピックアップカウンター付近でふ')
##引数のURLにHTTP接続します
##失敗かどうかは呼出側で要判定
def _request_once(self, url: Final[str]) -> Response:
if self._proxy_is_needed:
res: Response = requests.get(url, timeout=self.REQUEST_TIMEOUT, headers=self.HEADERS, allow_redirects=False, proxies=self.PROXIES)
else:
res: Response = requests.get(url, timeout=self.REQUEST_TIMEOUT, headers=self.HEADERS, allow_redirects=False)
sleep(self.WAIT_TIME) ##DoS対策で待つ
return res
##HTTP接続を再試行回数まで試します
##成功すると結果を返します
##接続失敗が何度も起きるとNoneを返します
##呼出側で要None判定
def _request(self, url: Final[str]) -> Response | None:
counter: int = 1 ##リクエスト挑戦回数を記録
while True:
try:
res: Response = 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 _check_slash(self) -> None | NoReturn:
if self.NITTER_INSTANCE[-1] != '/':
raise RuntimeError('NITTER_INSTANCEの末尾には/が必須です')
if self.ARCHIVE_TODAY[-1] != '/':
raise RuntimeError('ARCHIVE_TODAYの末尾には/が必須です')
if self.ARCHIVE_TODAY_STANDARD[-1] != '/':
raise RuntimeError('ARCHIVE_TODAY_STANDARDの末尾には/が必須です')
if self.TWITTER_URL[-1] != '/':
raise RuntimeError('TWITTER_URLの末尾には/が必須です')
##Ｔorが使えているかチェック
def _check_tor_proxy_is_needed(self) -> bool | NoReturn:
initial_proxy_is_needed: bool = self._proxy_is_needed
print('Ｔorのチェック中ですを')
# プロキシなしでTorにアクセスできるかどうか
self._proxy_is_needed = False
res: Response = self._request_once('https://check.torproject.org/api/ip') ##リクエスト
is_tor: bool = json.loads(res.text)['IsTor']
if is_tor:
print('Ｔor connection OK')
self._proxy_is_needed = initial_proxy_is_needed
return False
# プロキシありでTorにアクセスできるかどうか
self._proxy_is_needed = True
try:
res = self._request_once('https://check.torproject.org/api/ip') ##リクエスト
is_tor = json.loads(res.text)['IsTor']
if is_tor:
print('Ｔor proxy OK')
self._proxy_is_needed = initial_proxy_is_needed
return True
else:
raise RuntimeError('サイトにＴorのIPでアクセスできていないなりを')
except requests.exceptions.ConnectionError as e:
print(e, file=sys.stderr)
print('通信がＴorのSOCKS proxyを経由していないなりを', file=sys.stderr)
exit(1)
##nitterのインスタンスが生きているかチェック
##死んでいたらそこで終了
##接続を一回しか試さない_request_onceを使っているのは
##激重インスタンスが指定されたとき試行回数増やして偶然成功してそのまま実行されるのを躱すため
def _check_nitter_instance(self) -> None | NoReturn:
print("Nitterのインスタンスチェック中ですを")
try:
res: Final[Response] = self._request_once(self.NITTER_INSTANCE) ##リクエスト
res.raise_for_status() ##HTTPステータスコードが200番台以外でエラー発生
except requests.exceptions.RequestException as e: ##エラー発生時は終了
print(e, file=sys.stderr)
print('インスタンスが死んでますを', file=sys.stderr)
exit(1)
print("Nitter OK")
##archive.todayのTor用インスタンスが生きているかチェック
def _check_archive_instance(self) -> None | NoReturn:
print("archive.todayのＴorインスタンスチェック中ですを")
try:
res: Final[Response] = self._request_once(self.ARCHIVE_TODAY) ##リクエスト
res.raise_for_status() ##HTTPステータスコードが200番台以外でエラー発生
except requests.exceptions.RequestException as e: ##エラー発生時は終了
print(e, file=sys.stderr)
print('インスタンスが死んでますを', file=sys.stderr)
exit(1)
print("archive.today OK")
##Invidiousのインスタンスのタプルを取得
def _invidious_instances(self) -> tuple[str] | NoReturn:
print("Invidiousのインスタンスリストを取得中ですを")
invidious_json: Response | None = self._request('https://api.invidious.io/instances.json')
if invidious_json is None:
print("Invidiousが死んでますを")
exit(1)
instance_list: list[str] = []
for instance_info in json.loads(invidious_json.text):
instance_list.append(instance_info[0])
# よく使われているものはチェック
if 'piped.kavin.rocks' not in instance_list:
instance_list.append('piped.kavin.rocks')
if 'piped.video' not in instance_list:
instance_list.append('piped.video')
return tuple(instance_list)
##ツイート収集するユーザー名を取得
##何も入力しないと尊師を指定するよう改良
def _get_name(self) -> str | NoReturn:
while True:
print('アカウント名を入れなければない。空白だと自動的に' + self.CALLINSHOW + 'になりますを')
account_str: str = input() ##ユーザー入力受付
##空欄で降臨ショー
if account_str == '':
return self.CALLINSHOW
else:
res: Response | None = self._request(urljoin(self.NITTER_INSTANCE, account_str)) ##リクエストして結果取得
if res is None : ##リクエスト失敗判定
self._fail()
soup: BeautifulSoup = BeautifulSoup(res.text, 'html.parser') ##beautifulsoupでレスポンス解析
if soup.title == self.NITTER_ERROR_TITLE: ##タイトルがエラーでないか判定
print(account_str + "は実在の人物ではありませんでした") ##エラー時ループに戻る
else:
print("最終的に出会ったのが@" + account_str + "だった。")
return account_str ##成功時アカウント名返す
##検索クエリを取得
def _get_query(self, name: Final[str]) -> Response | NoReturn:
while True:
print("検索クエリを入れるナリ。複数ある時は一つの塊ごとに入力するナリ。空欄を入れると検索開始ナリ。")
print("例：「無能」→改行→「脱糞」→改行→「弟殺し」→改行→空欄で改行")
query_str: list[str] = [name] ##検索クエリ文字列をquery_strに収納。ユーザー名を先に含めておく
query_input: str = input() ##ユーザー入力受付
##空欄が押されるまでユーザー入力受付
while query_input != '':
query_str.append(quote(query_input))
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("適切なクエリを入力することを切に望む。")
##接続失敗時処理
def _fail(self) -> NoReturn:
print("接続失敗しすぎで強制終了ナリ")
if len(self._txt_data) > 0: ##取得成功したデータがあれば発行
print("取得成功した分だけ発行しますを")
self._make_txt()
else:
exit(1) ##終了
##self._txt_dataにwikiでテーブル表示にするためのタグをつける
def _convert_to_text_table(self, text) -> str:
return '{|class="wikitable" style="text-align: left;"\n' + text + '|}'
##テキスト発行
def _make_txt(self) -> NoReturn:
self._next_day()
result_txt: Final[str] = '\n'.join(self._txt_data) ##リストを合体
##ファイル出力
with codecs.open('tweet.txt', 'w', 'utf-8') as f:
f.write(result_txt)
print("テキストファイル手に入ったやで〜")
exit(0) ##終了
##記録を中断するツイート
def _stop_word(self) -> str:
print(f"ここにツイートの本文を入れる！すると・・・・なんとそれを含むツイートで記録を中断する！（これは本当の仕様です。）ちなみに、空欄だと{self.LIMIT_N_TWEETS}件まで終了しない。")
end_str: Final[str] = input() ##ユーザー入力受付
return end_str
##Twitterの画像をダウンロード
def _download_media(self, media_name: Final[str]) -> bool:
os.makedirs(self.MEDIA_DIR, exist_ok=True)
url: Final[str] = urljoin('https://pbs.twimg.com/media/', media_name)
res: Final[Response | None] = self._request(url)
if res is not None:
if 'image' not in res.headers['content-type']:
return False
with open(os.path.join(self.MEDIA_DIR, media_name), "wb") as f:
f.write(res.content)
return True
else:
return False
##ツイートの日付を取得
def _tweet_date(self, tweet: bs4.element.Tag) -> datetime:
date_str: str = tweet.find(class_='tweet-date').a['title']
date: datetime = datetime.strptime(date_str, '%b %d, %Y · %I:%M %p %Z').replace(tzinfo=ZoneInfo('UTC')).astimezone(ZoneInfo('Asia/Tokyo'))
return date
#self._dateの日付のツイートがなくなったときの処理
def _next_day(self, date: datetime | None = None) -> None:
if self._txt_data[0]: # 空でなければ出力
self._txt_data[0] = self._convert_to_text_table(self._txt_data[0])
if os.name == 'nt': # Windows
self._txt_data[0] = self._date.strftime('\n=== %#m月%#d日 ===\n') + self._txt_data[0]
print(self._date.strftime('%#m月%#d日のツイートを取得完了ですを'))
else: # Mac or Linux
self._txt_data[0] = self._date.strftime('\n=== %-m月%-d日 ===\n') + self._txt_data[0]
print(self._date.strftime('%-m月%-d日のツイートを取得完了ですを'))
if date is not None:
self._txt_data.insert(0, '')
self._date = date
def _get_tweet_media(self, tweet: bs4.element.Tag) -> str:
tweet_media: bs4.element.Tag | None = tweet.select_one('.tweet-body > .attachments') # 引用リツイート内のメディアを選択しないように.tweet-body直下の.attachmentsを選択
media_txt: str = ''
if tweet_media is not None:
media_list: list[str] = []
# ツイートの画像の取得
for image_a in tweet_media.select('.attachment.image a'):
try:
media_name: str = [group for group in re.search(r'%2F([^%]*\.jpg)|%2F([^%]*\.jpeg)|%2F([^%]*\.png)|%2F([^%]*\.gif)', image_a.get('href')).groups() if group][0]
media_list.append(f"[[ファイル:{media_name}|240px]]")
if self._download_media(media_name):
print(os.path.join(self.MEDIA_DIR, media_name) + ' をアップロードしなければない。')
else:
print(urljoin('https://pbs.twimg.com/media/', media_name) + ' をアップロードしなければない。')
except AttributeError as e:
tweet_url: str = urljoin(self.TWITTER_URL, re.sub('#[^#]*$', '', tweet.find(class_='tweet-link').get('href'))) ##ツイートのURL作成
print(f"{tweet_url}の画像が取得できませんでしたを 当職無能")
media_list.append(f"[[ファイル:(画像の取得ができませんでした)|240px]]")
# ツイートの動画の取得。ffmpegが入っていなければ手動で取得すること
for i, video in enumerate(tweet_media.select('.attachment.video-container video')):
tweet_url: str = urljoin(self.TWITTER_URL, re.sub('#[^#]*$', '', tweet.find(class_='tweet-link').get('href'))) ##ツイートのURL作成
if subprocess.run(['which', 'ffmpeg'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL).returncode != 0:
print(f"ffmpegがないため{tweet_url}の動画が取得できませんでしたを")
media_list.append(f"[[ファイル:(動画の取得ができませんでした)|240px]]")
else: # ffmpegがある場合
media_url: str = unquote(re.search(r'[^\/]+$', video.get('data-url')).group(0))
tweet_id: str = tweet_url.split('/')[-1]
# 動画のダウンロード
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
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
# 取得成功したら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
if ts2mp4_returncode == 0:
print(f"{os.path.join(self.MEDIA_DIR, tweet_id)}_{i}.mp4をアップロードしなければない。")
else:
print(f"{os.path.join(self.MEDIA_DIR, tweet_id)}_{i}.tsをmp4に変換してアップロードしなければない。")
media_list.append(f"[[ファイル:{tweet_id}_{i}.mp4|240px]]")
else:
print(f"{tweet_url}の動画が取得できませんでしたを 当職無能")
media_list.append(f"[[ファイル:(動画の取得ができませんでした)|240px]]")
media_txt = '<br>\n' + ' '.join(media_list)
return media_txt
def _get_tweet_quote(self, tweet: bs4.element.Tag) -> str:
tweet_quote: Final[bs4.element.Tag | None] = tweet.select_one('.tweet-body > .quote.quote-big') # 引用リツイートを選択
quote_txt: str = ''
if tweet_quote is not None:
link: str = tweet_quote.select_one('.quote-link').get('href')
link = re.sub('#.*$', '', link)
link = urljoin(self.TWITTER_URL, link)
quote_txt = '<br>\n' + self._archive_url(link, link)
return quote_txt
def _get_tweet_poll(self, tweet: bs4.element.Tag) -> str:
tweet_poll: Final[bs4.element.Tag | None] = tweet.select_one('.tweet-body > .poll')
poll_txt: str = ''
if tweet_poll is not None:
poll_meters: Final[bs4.element.ResultSet] = tweet_poll.select('.poll-meter')
poll_txt += '<br>\n'
for poll_meter in poll_meters:
ratio: str = poll_meter.select_one('.poll-choice-value').text
if 'leader' in poll_meter['class']:
poll_txt += f'<br>\n <span style="display: inline-block; width: 30em; background: linear-gradient(to right, rgba(29, 155, 240, 0.58) 0 {ratio}, transparent {ratio} 100%); font-weight: bold;">' + ratio + ' ' + poll_meter.select_one('.poll-choice-option').text + '</span>'
else:
poll_txt += f'<br>\n <span style="display: inline-block; width: 30em; background: linear-gradient(to right, rgb(207, 217, 222) 0 {ratio}, transparent {ratio} 100%);">' + ratio + ' ' + poll_meter.select_one('.poll-choice-option').text + '</span>'
poll_txt += '<br>\n <span style="font-size: small;">' + tweet_poll.select_one('.poll-info').text + '</span>'
return poll_txt
def _check_additional_info(self, tweet: bs4.element.Tag, tweet_url: str) -> None:
if '/i/web/status/' in tweet.text:
print(f"{tweet_url}をには編集前のツイートか省略が存在する可能性があります")
#一ツイートのブロックごとにリストで取得。そのままtimeline-itemクラスをfind_allするとツイートの順番が逆転するので、順番通りに取得するよう処理
def _get_timeline_items(self, soup: BeautifulSoup) -> list[bs4.element.Tag]:
timeline_item_list: list[bs4.element.Tag] = []
for item_or_list in soup.select('.timeline > .timeline-item, .timeline > .thread-line'):
if 'thread-line' in item_or_list.attrs['class']:
for item in reversed(item_or_list.select('.timeline-item')):
timeline_item_list.append(item)
else:
timeline_item_list.append(item_or_list)
return timeline_item_list
#ページからツイート本文をself._txt_dataに収めていく
def get_tweet(self) -> None | NoReturn:
soup: Final[BeautifulSoup] = BeautifulSoup(self._page.text, 'html.parser') ##beautifulsoupでレスポンス解析
tweets: Final[list[bs4.element.Tag]] = self._get_timeline_items(soup) ##一ツイートのブロックごとにリストで取得
for tweet in tweets: ##一ツイート毎に処理
if tweet.a.text == self.NEWEST: ##Load Newestのボタンは処理しない
continue
if tweet.find(class_='retweet-header') is not None: ##retweet-headerはリツイートを示すので入っていれば処理しない
continue
tweet_url: str = urljoin(self.TWITTER_URL, re.sub('#[^#]*$', '', tweet.find(class_='tweet-link').get('href'))) ##ツイートのURL作成
date: datetime = self._tweet_date(tweet)
if date.year != self._date.year or date.month != self._date.month or date.day != self._date.day:
self._next_day(date)
archived_tweet_url: str = self._callinshowlink_url(tweet_url) ##ツイートURLをテンプレートCallinShowlinkに変化
tweet_content: bs4.element.Tag = tweet.find(class_='tweet-content media-body') ##ツイートの中身だけ取り出す
self._archive_soup(tweet_content) ##ツイートの中身のリンクをテンプレートArchiveに変化
poll_txt: str = self._get_tweet_poll(tweet) ##投票の取得
media_txt: str = self._get_tweet_media(tweet) ##ツイートに画像などのメディアを追加
quote_txt: str = self._get_tweet_quote(tweet) ##引用リツイートの場合、元ツイートを追加
self._check_additional_info(tweet, tweet_url) ##編集や長いツイートの省略をチェック
self._txt_data[0] = '!' + archived_tweet_url + '\n|-\n|\n' \
+ self._escape_wiki_reserved_words(tweet_content.get_text()) \
+ poll_txt + quote_txt + media_txt + '\n|-\n' \
+ self._txt_data[0] ##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()
#MediaWiki文法と衝突する文字を無効化する
def _escape_wiki_reserved_words(self, text: str) -> str:
text = text.replace('\n', '<br>\n')
text = re.sub(r'^ ', ' ', 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)
return text
#tagをテンプレートArchiveの文字列に変化させる
def _archive_soup(self, tag: bs4.element.Tag) -> None:
urls_in_tweet: Final[bs4.element.ResultSet] = tag.find_all('a')
for url in urls_in_tweet:
if url.get('href').startswith('https://') or url.get('href').startswith('http://'): ##先頭にhttpが付いていない物はハッシュタグの検索ページへのリンクなので処理しない
if url.get('href').startswith('https' + self.NITTER_INSTANCE[4:]):
#Nitter上のTwitterへのリンクを直す
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に変化
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に変化
elif self._invidious_pattern.search(url.get('href')):
#Nitter上のYouTubeへのリンクをInvidiousのものから直す
url_link: str = url.get('href')
if re.match('https://[^/]+/[^/]+/', 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に変化
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に変化
else:
url.replace_with(self._archive_url(url.get('href'), url.get('href'))) ##テンプレートArchiveに変化
#URLをテンプレートArchiveに変化させる
def _archive_url(self, url: Final[str], text: Final[str]) -> str:
return '{{Archive|1=' + unquote(url) + '|2=' + self._archive(url) + '|3=' + unquote(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:
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使って結果を取得
if res is None : ##魚拓接続失敗時処理
print(archive_url + 'にアクセス失敗ナリ。出力されるテキストにはそのまま記載されるナリ。')
else:
soup: Final[BeautifulSoup] = BeautifulSoup(res.text, 'html.parser') ##beautifulsoupでレスポンス解析
content: bs4.element.Tag = soup.find(id="CONTENT") ##archive.todayの魚拓一覧ページの中身だけ取得
if content is None or 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) -> None | NoReturn:
soup: Final[BeautifulSoup] = BeautifulSoup(self._page.text, 'html.parser') ##beautifulsoupでレスポンス解析
show_mores: Final[bs4.element.ResultSet] = soup.find_all(class_="show-more")
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頭部分と合体
res: Final[Response | None] = self._request(new_url) ##接続してHTML取ってくる
if res is None:
self._fail()
new_page_soup: Final[BeautifulSoup] = BeautifulSoup(res.text, 'html.parser') ##beautifulsoupでレスポンス解析
if new_page_soup.find(class_="timeline-end") is None: ##ツイートの終端ではtimeline-endだけのページになるので判定
print(res.url + 'に移動しますを')
self._page = res ##まだ残りツイートがあるのでページを返して再度ツイート本文収集
else:
print("急に残りツイートが無くなったな終了するか")
self._make_txt()
if __name__ == '__main__':
if sys.version_info.major < 3 or (sys.version_info.major == 3 and sys.version_info.minor < 10):
print('Pythonのバージョンを3.10以上に上げて下さい', file=sys.stderr)
exit(1)
krsw: Final[bool] = len(sys.argv) > 1 and sys.argv[1] == 'krsw' ##コマンドライン引数があるかどうかのフラグ
twitter_archiver: TwitterArchiver = TwitterArchiver(krsw)
##ツイートを取得し終えるまでループ
while True:
twitter_archiver.get_tweet() ##self._txt_dataにページ内のリツイート以外の全ツイートの中身突っ込んでいく
twitter_archiver.go_to_new_page() ##新しいページ取得
実行例
20件での実行例。
12月10日
| https://twitter.com/CallinShow/status/1601539154256744449(魚拓) |
|---|
|
なお、ひめかファンのワタクシ、ブロックされております。 |
| https://twitter.com/CallinShow/status/1601542511302160384(魚拓) |
|
今日もお疲れ。 |
| https://twitter.com/CallinShow/status/1601569138379718656(魚拓) |
|
https://youtu.be/QR6Gj0MKcew(魚拓) |
| https://twitter.com/CallinShow/status/1601572951463428096(魚拓) |
|
菊地翔 |
| https://twitter.com/CallinShow/status/1601588268487041024(魚拓) |
|
日曜阪神11R 阪神JF |
12月11日
12月12日
| https://twitter.com/CallinShow/status/1601955353792430081(魚拓) |
|---|
|
エクシア合同会社のツイートの中に、岡ちゃんのツイートをリツイートしてしまってごめん。 |
| https://twitter.com/CallinShow/status/1601958600259371011(魚拓) |
|
日曜日に誰もエクシア合同会社のことなんか、呟きたいなんて思わないのが普通だと思う。 |
| https://twitter.com/CallinShow/status/1601959896252981249(魚拓) |
|
ときにはふざけたツイートをしているように思うかもしれない。 |
| https://twitter.com/CallinShow/status/1601961096528613376(魚拓) |
|
エクシア合同会社と闘う大人たちがやっていることは、Twitterを使った世直し運動なんだ。 |